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

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

本文来自:Segmentfault

感谢作者:兄弟连区块链

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

• 编辑
• 预览