区块链教程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)

//获取怀疑脱机节点的只读通道

PresumedDead()

//关闭到某个节点的连接

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

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