区块链教程Fabric1.0源代码分析流言算法Gossip服务端一

区块链教程Fabric1.0源代码分析流言算法Gossip服务端一，2018年下半年，区块链行业正逐渐褪去发展之初的浮躁、回归理性，表面上看相关人才需求与身价似乎正在回落。但事实上，正是初期泡沫的渐退，让人们更多的关注点放在了区块链真正的技术之上。

Fabric 1.0源代码笔记 之 gossip（流言算法） #GossipServer（Gossip服务端）

1、GossipServer概述

GossipServer相关代码，分布在protos/gossip、gossip/comm目录下。目录结构如下：

• protos/gossip目录：

* message.pb.go，GossipClient接口定义及实现，GossipServer接口定义。

• gossip/comm目录：

* comm.go，Comm接口定义。

* conn.go，connFactory接口定义，以及connectionStore结构体及方法。

* comm_impl.go，commImpl结构体及方法（同时实现GossipServer接口/Comm接口/connFactory接口）。

* demux.go，ChannelDeMultiplexer结构体及方法。

2、GossipClient接口定义及实现

2.1、GossipClient接口定义

type GossipClient interface {
    // GossipStream is the gRPC stream used for sending and receiving messages
    GossipStream(ctx context.Context, opts ...grpc.CallOption) (Gossip_GossipStreamClient, error)
    // Ping is used to probe a remote peer's aliveness
    Ping(ctx context.Context, in *Empty, opts ...grpc.CallOption) (*Empty, error)
}
//代码在protos/gossip/message.pb.go

2.2、GossipClient接口实现

type gossipClient struct {
    cc *grpc.ClientConn
}

func NewGossipClient(cc *grpc.ClientConn) GossipClient {
    return &gossipClient{cc}
}

func (c *gossipClient) GossipStream(ctx context.Context, opts ...grpc.CallOption) (Gossip_GossipStreamClient, error) {
    stream, err := grpc.NewClientStream(ctx, &_Gossip_serviceDesc.Streams[0], c.cc, "/gossip.Gossip/GossipStream", opts...)
    if err != nil {
        return nil, err
    }
    x := &gossipGossipStreamClient{stream}
    return x, nil
}

func (c *gossipClient) Ping(ctx context.Context, in *Empty, opts ...grpc.CallOption) (*Empty, error) {
    out := new(Empty)
    err := grpc.Invoke(ctx, "/gossip.Gossip/Ping", in, out, c.cc, opts...)
    if err != nil {
        return nil, err
    }
    return out, nil
}
//代码在protos/gossip/message.pb.go

2.3、Gossip_GossipStreamClient接口定义及实现

type Gossip_GossipStreamClient interface {
    Send(*Envelope) error
    Recv() (*Envelope, error)
    grpc.ClientStream
}

type gossipGossipStreamClient struct {
    grpc.ClientStream
}

func (x *gossipGossipStreamClient) Send(m *Envelope) error {
    return x.ClientStream.SendMsg(m)
}

func (x *gossipGossipStreamClient) Recv() (*Envelope, error) {
    m := new(Envelope)
    if err := x.ClientStream.RecvMsg(m); err != nil {
        return nil, err
    }
    return m, nil
}
//代码在protos/gossip/message.pb.go

3、GossipServer接口定义

3.1、GossipServer接口定义

type GossipServer interface {
    // GossipStream is the gRPC stream used for sending and receiving messages
    GossipStream(Gossip_GossipStreamServer) error
    // Ping is used to probe a remote peer's aliveness
    Ping(context.Context, *Empty) (*Empty, error)
}

func RegisterGossipServer(s *grpc.Server, srv GossipServer) {
    s.RegisterService(&_Gossip_serviceDesc, srv)
}

func _Gossip_GossipStream_Handler(srv interface{}, stream grpc.ServerStream) error {
    return srv.(GossipServer).GossipStream(&gossipGossipStreamServer{stream})
}

func _Gossip_Ping_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) {
    in := new(Empty)
    if err := dec(in); err != nil {
        return nil, err
    }
    if interceptor == nil {
        return srv.(GossipServer).Ping(ctx, in)
    }
    info := &grpc.UnaryServerInfo{
        Server:     srv,
        FullMethod: "/gossip.Gossip/Ping",
    }
    handler := func(ctx context.Context, req interface{}) (interface{}, error) {
        return srv.(GossipServer).Ping(ctx, req.(*Empty))
    }
    return interceptor(ctx, in, info, handler)
}

var _Gossip_serviceDesc = grpc.ServiceDesc{
    ServiceName: "gossip.Gossip",
    HandlerType: (*GossipServer)(nil),
    Methods: []grpc.MethodDesc{
        {
            MethodName: "Ping",
            Handler:    _Gossip_Ping_Handler,
        },
    },
    Streams: []grpc.StreamDesc{
        {
            StreamName:    "GossipStream",
            Handler:       _Gossip_GossipStream_Handler,
            ServerStreams: true,
            ClientStreams: true,
        },
    },
    Metadata: "gossip/message.proto",
}
//代码在protos/gossip/message.pb.go

3.2、Gossip_GossipStreamServer接口定义及实现

type Gossip_GossipStreamServer interface {
    Send(*Envelope) error
    Recv() (*Envelope, error)
    grpc.ServerStream
}

type gossipGossipStreamServer struct {
    grpc.ServerStream
}

func (x *gossipGossipStreamServer) Send(m *Envelope) error {
    return x.ServerStream.SendMsg(m)
}

func (x *gossipGossipStreamServer) Recv() (*Envelope, error) {
    m := new(Envelope)
    if err := x.ServerStream.RecvMsg(m); err != nil {
        return nil, err
    }
    return m, nil
}
//代码在protos/gossip/message.pb.go

4、Comm接口/connFactory接口定义

4.1、Comm接口定义

type Comm interface {
    //返回此实例的 PKI id
    GetPKIid() common.PKIidType
    //向节点发送消息
    Send(msg *proto.SignedGossipMessage, peers ...*RemotePeer)
    //探测远程节点是否有响应
    Probe(peer *RemotePeer) error
    //握手验证远程节点
    Handshake(peer *RemotePeer) (api.PeerIdentityType, error)
    Accept(common.MessageAcceptor) <-chan proto.ReceivedMessage
    //获取怀疑脱机节点的只读通道
    PresumedDead() <-chan common.PKIidType
    //关闭到某个节点的连接
    CloseConn(peer *RemotePeer)
    //关闭
    Stop()
}
//代码在gossip/comm/comm.go

4.2、connFactory接口定义

type connFactory interface {
    createConnection(endpoint string, pkiID common.PKIidType) (*connection, error)
}
//代码在gossip/comm/conn.go

5、commImpl结构体及方法（同时实现GossipServer接口/Comm接口/connFactory接口）

5.1、commImpl结构体定义

type commImpl struct {
    selfCertHash   []byte
    peerIdentity   api.PeerIdentityType
    idMapper       identity.Mapper
    logger         *logging.Logger
    opts           []grpc.DialOption
    secureDialOpts func() []grpc.DialOption
    connStore      *connectionStore
    PKIID          []byte
    deadEndpoints  chan common.PKIidType
    msgPublisher   *ChannelDeMultiplexer
    lock           *sync.RWMutex
    lsnr           net.Listener
    gSrv           *grpc.Server
    exitChan       chan struct{}
    stopWG         sync.WaitGroup
    subscriptions  []chan proto.ReceivedMessage
    port           int
    stopping       int32
}
//代码在gossip/comm/comm_impl.go

未完待续欢迎继续关注兄弟连区块链教程分享！