哪家做网站做得好,网页开发哪家好,网站开发 学习步骤,公司营销网络怎么写区块链教程Fabric1.0源代码分析流言算法Gossip服务端二 Fabric 1.0源代码笔记 之 gossip#xff08;流言算法#xff09; #GossipServer#xff08;Gossip服务端#xff09; 5.2、commImpl结构体方法 //conn.serviceConnection()#xff0c;启动连接服务
func (c *commImp… 区块链教程Fabric1.0源代码分析流言算法Gossip服务端二 Fabric 1.0源代码笔记 之 gossip流言算法 #GossipServerGossip服务端 5.2、commImpl结构体方法 //conn.serviceConnection()启动连接服务
func (c *commImpl) GossipStream(stream proto.Gossip_GossipStreamServer) error
//return proto.Empty{}
func (c *commImpl) Ping(context.Context, *proto.Empty) (*proto.Empty, error)func (c *commImpl) GetPKIid() common.PKIidType
//向指定节点发送消息
func (c *commImpl) Send(msg *proto.SignedGossipMessage, peers ...*RemotePeer)
//探测远程节点是否有响应_, err cl.Ping(context.Background(), proto.Empty{})
func (c *commImpl) Probe(remotePeer *RemotePeer) error
//握手验证远程节点_, err cl.Ping(context.Background(), proto.Empty{})
func (c *commImpl) Handshake(remotePeer *RemotePeer) (api.PeerIdentityType, error)
func (c *commImpl) Accept(acceptor common.MessageAcceptor) -chan proto.ReceivedMessage
func (c *commImpl) PresumedDead() -chan common.PKIidType
func (c *commImpl) CloseConn(peer *RemotePeer)
func (c *commImpl) Stop()//创建并启动gRPC Server以及注册GossipServer实例
func NewCommInstanceWithServer(port int, idMapper identity.Mapper, peerIdentity api.PeerIdentityType,
//将GossipServer实例注册至peerServer
func NewCommInstance(s *grpc.Server, cert *tls.Certificate, idStore identity.Mapper,
func extractRemoteAddress(stream stream) string
func readWithTimeout(stream interface{}, timeout time.Duration, address string) (*proto.SignedGossipMessage, error)
//创建gRPC Servergrpc.NewServer(serverOpts...)
func createGRPCLayer(port int) (*grpc.Server, net.Listener, api.PeerSecureDialOpts, []byte)//创建与服务端连接
func (c *commImpl) createConnection(endpoint string, expectedPKIID common.PKIidType) (*connection, error)
//向指定节点发送消息
func (c *commImpl) sendToEndpoint(peer *RemotePeer, msg *proto.SignedGossipMessage)
//return atomic.LoadInt32(c.stopping) int32(1)
func (c *commImpl) isStopping() bool
func (c *commImpl) emptySubscriptions()
func (c *commImpl) authenticateRemotePeer(stream stream) (*proto.ConnectionInfo, error)
func (c *commImpl) disconnect(pkiID common.PKIidType)
func (c *commImpl) createConnectionMsg(pkiID common.PKIidType, certHash []byte, cert api.PeerIdentityType, signer proto.Signer) (*proto.SignedGossipMessage, error)
//代码在gossip/comm/comm_impl.go 5.2.1、func NewCommInstanceWithServer(port int, idMapper identity.Mapper, peerIdentity api.PeerIdentityType,secureDialOpts api.PeerSecureDialOpts, dialOpts ...grpc.DialOption) (Comm, error) 创建并启动gRPC Server以及注册GossipServer实例 func NewCommInstanceWithServer(port int, idMapper identity.Mapper, peerIdentity api.PeerIdentityType,secureDialOpts api.PeerSecureDialOpts, dialOpts ...grpc.DialOption) (Comm, error) {var ll net.Listenervar s *grpc.Servervar certHash []byteif len(dialOpts) 0 {//peer.gossip.dialTimeoutgRPC连接拨号的超时dialOpts []grpc.DialOption{grpc.WithTimeout(util.GetDurationOrDefault(peer.gossip.dialTimeout, defDialTimeout))}}if port 0 {//创建gRPC Servergrpc.NewServer(serverOpts...)s, ll, secureDialOpts, certHash createGRPCLayer(port)}commInst : commImpl{selfCertHash: certHash,PKIID: idMapper.GetPKIidOfCert(peerIdentity),idMapper: idMapper,logger: util.GetLogger(util.LoggingCommModule, fmt.Sprintf(%d, port)),peerIdentity: peerIdentity,opts: dialOpts,secureDialOpts: secureDialOpts,port: port,lsnr: ll,gSrv: s,msgPublisher: NewChannelDemultiplexer(),lock: sync.RWMutex{},deadEndpoints: make(chan common.PKIidType, 100),stopping: int32(0),exitChan: make(chan struct{}, 1),subscriptions: make([]chan proto.ReceivedMessage, 0),}commInst.connStore newConnStore(commInst, commInst.logger)if port 0 {commInst.stopWG.Add(1)go func() {defer commInst.stopWG.Done()s.Serve(ll) //启动gRPC Server}()//commInst注册到gRPC Serverproto.RegisterGossipServer(s, commInst)}return commInst, nil
}//代码在gossip/comm/comm_impl.go 5.2.2、func NewCommInstance(s grpc.Server, cert tls.Certificate, idStore identity.Mapper,peerIdentity api.PeerIdentityType, secureDialOpts api.PeerSecureDialOpts,dialOpts ...grpc.DialOption) (Comm, error) 将GossipServer实例注册至peerServer func NewCommInstance(s *grpc.Server, cert *tls.Certificate, idStore identity.Mapper,peerIdentity api.PeerIdentityType, secureDialOpts api.PeerSecureDialOpts,dialOpts ...grpc.DialOption) (Comm, error) {dialOpts append(dialOpts, grpc.WithTimeout(util.GetDurationOrDefault(peer.gossip.dialTimeout, defDialTimeout)))//构造commImplcommInst, err : NewCommInstanceWithServer(-1, idStore, peerIdentity, secureDialOpts, dialOpts...)if cert ! nil {inst : commInst.(*commImpl)inst.selfCertHash certHashFromRawCert(cert.Certificate[0])}proto.RegisterGossipServer(s, commInst.(*commImpl))return commInst, nil
}//代码在gossip/comm/comm_impl.go //创建与服务端连接 5.2.3、func (c commImpl) createConnection(endpoint string, expectedPKIID common.PKIidType) (connection, error) func (c *commImpl) createConnection(endpoint string, expectedPKIID common.PKIidType) (*connection, error) {var err errorvar cc *grpc.ClientConnvar stream proto.Gossip_GossipStreamClientvar pkiID common.PKIidTypevar connInfo *proto.ConnectionInfovar dialOpts []grpc.DialOptiondialOpts append(dialOpts, c.secureDialOpts()...)dialOpts append(dialOpts, grpc.WithBlock())dialOpts append(dialOpts, c.opts...)cc, err grpc.Dial(endpoint, dialOpts...)cl : proto.NewGossipClient(cc)if _, err cl.Ping(context.Background(), proto.Empty{}); err ! nil {cc.Close()return nil, err}ctx, cf : context.WithCancel(context.Background())stream, err cl.GossipStream(ctx)connInfo, err c.authenticateRemotePeer(stream)pkiID connInfo.IDconn : newConnection(cl, cc, stream, nil)conn.pkiID pkiIDconn.info connInfoconn.logger c.loggerconn.cancel cfh : func(m *proto.SignedGossipMessage) {c.msgPublisher.DeMultiplex(ReceivedMessageImpl{conn: conn,lock: conn,SignedGossipMessage: m,connInfo: connInfo,})}conn.handler hreturn conn, nil
}
//代码在gossip/comm/comm_impl.go 6、connectionStore和connection结构体及方法 6.1、connection结构体及方法 type connection struct {cancel context.CancelFuncinfo *proto.ConnectionInfooutBuff chan *msgSendinglogger *logging.Logger // loggerpkiID common.PKIidType // pkiID of the remote endpointhandler handler // function to invoke upon a message receptionconn *grpc.ClientConn // gRPC connection to remote endpointcl proto.GossipClient // gRPC stub of remote endpointclientStream proto.Gossip_GossipStreamClient // client-side stream to remote endpointserverStream proto.Gossip_GossipStreamServer // server-side stream to remote endpointstopFlag int32 // indicates whether this connection is in process of stoppingstopChan chan struct{} // a method to stop the server-side gRPC call from a different go-routinesync.RWMutex // synchronizes access to shared variables
}//构造connection
func newConnection(cl proto.GossipClient, c *grpc.ClientConn, cs proto.Gossip_GossipStreamClient, ss proto.Gossip_GossipStreamServer) *connection
//关闭connection
func (conn *connection) close()
//atomic.LoadInt32((conn.stopFlag)) int32(1)
func (conn *connection) toDie() bool
//conn.outBuff - m其中m为msgSending{envelope: msg.Envelope,onErr: onErr,}
func (conn *connection) send(msg *proto.SignedGossipMessage, onErr func(error))
//go conn.readFromStream(errChan, msgChan)、go conn.writeToStream()同时msg : -msgChanconn.handler(msg)
func (conn *connection) serviceConnection() error
//循环不间断从conn.outBuff取数据然后stream.Send(m.envelope)
func (conn *connection) writeToStream()
//循环不间断envelope, err : stream.Recv()、msg, err : envelope.ToGossipMessage()、msgChan - msg
func (conn *connection) readFromStream(errChan chan error, msgChan chan *proto.SignedGossipMessage)
//获取conn.serverStream
func (conn *connection) getStream() stream
//代码在gossip/comm/conn.go 6.2、connectionStore结构体及方法 type connectionStore struct {logger *logging.Logger // loggerisClosing bool // whether this connection store is shutting downconnFactory connFactory // creates a connection to remote peersync.RWMutex // synchronize access to shared variablespki2Conn map[string]*connection //connection map, key为pkiIDvalue为connectiondestinationLocks map[string]*sync.RWMutex //mapping between pkiIDs and locks,// used to prevent concurrent connection establishment to the same remote endpoint
}//构造connectionStore
func newConnStore(connFactory connFactory, logger *logging.Logger) *connectionStore
//从connection map中获取连接如无则创建并启动连接并写入connection map中
func (cs *connectionStore) getConnection(peer *RemotePeer) (*connection, error)
//连接数量
func (cs *connectionStore) connNum() int
//关闭指定连接
func (cs *connectionStore) closeConn(peer *RemotePeer)
//关闭所有连接
func (cs *connectionStore) shutdown()
func (cs *connectionStore) onConnected(serverStream proto.Gossip_GossipStreamServer, connInfo *proto.ConnectionInfo) *connection
//注册连接
func (cs *connectionStore) registerConn(connInfo *proto.ConnectionInfo, serverStream proto.Gossip_GossipStreamServer) *connection
//关闭指定连接
func (cs *connectionStore) closeByPKIid(pkiID common.PKIidType)
//代码在gossip/comm/conn.go 6.2.1、func (cs connectionStore) getConnection(peer RemotePeer) (*connection, error) func (cs *connectionStore) getConnection(peer *RemotePeer) (*connection, error) {cs.RLock()isClosing : cs.isClosingcs.RUnlock()pkiID : peer.PKIIDendpoint : peer.Endpointcs.Lock()destinationLock, hasConnected : cs.destinationLocks[string(pkiID)]if !hasConnected {destinationLock sync.RWMutex{}cs.destinationLocks[string(pkiID)] destinationLock}cs.Unlock()destinationLock.Lock()cs.RLock()//从connection map中获取conn, exists : cs.pki2Conn[string(pkiID)]if exists {cs.RUnlock()destinationLock.Unlock()return conn, nil}cs.RUnlock()//创建连接createdConnection, err : cs.connFactory.createConnection(endpoint, pkiID)destinationLock.Unlock()conn createdConnectioncs.pki2Conn[string(createdConnection.pkiID)] conngo conn.serviceConnection() //启动连接的消息接收处理、以及向对方节点发送消息return conn, nil
}
//代码在gossip/comm/conn.go 7、ChannelDeMultiplexer结构体及方法多路复用器 type ChannelDeMultiplexer struct {channels []*channellock *sync.RWMutexclosed int32
}//构造ChannelDeMultiplexer
func NewChannelDemultiplexer() *ChannelDeMultiplexer
//atomic.LoadInt32(m.closed) int32(1)
func (m *ChannelDeMultiplexer) isClosed() bool
//关闭
func (m *ChannelDeMultiplexer) Close()
//添加通道
func (m *ChannelDeMultiplexer) AddChannel(predicate common.MessageAcceptor) chan interface{}
//挨个通道发送消息
func (m *ChannelDeMultiplexer) DeMultiplex(msg interface{}) 转载于:https://blog.51cto.com/14041296/2311323
