From d09b7dbbef0da3ef4713c6b573d87c0dbe4f9628 Mon Sep 17 00:00:00 2001 From: Milos Gajdos Date: Wed, 28 Aug 2019 20:11:19 +0100 Subject: [PATCH] Broadcast neighbourhood; fix critical bugs in channel connections This commit introduces neighbourhood announcements which allows to maintaing neighbour map if each next-hop node. It also fixes a critical bug when accepting connections for a particular tunnel channel. --- network/default.go | 371 +++++++++++++++++++++++++++++---- network/network.go | 4 +- network/proto/network.micro.go | 21 ++ network/proto/network.pb.go | 227 ++++++++++++++++++++ network/proto/network.proto | 31 +++ 5 files changed, 607 insertions(+), 47 deletions(-) create mode 100644 network/proto/network.micro.go create mode 100644 network/proto/network.pb.go create mode 100644 network/proto/network.proto diff --git a/network/default.go b/network/default.go index 8a612394..c903d4b2 100644 --- a/network/default.go +++ b/network/default.go @@ -7,9 +7,10 @@ import ( "github.com/golang/protobuf/proto" "github.com/micro/go-micro/client" rtr "github.com/micro/go-micro/client/selector/router" + pbNet "github.com/micro/go-micro/network/proto" "github.com/micro/go-micro/proxy" "github.com/micro/go-micro/router" - pb "github.com/micro/go-micro/router/proto" + pbRtr "github.com/micro/go-micro/router/proto" "github.com/micro/go-micro/server" "github.com/micro/go-micro/transport" "github.com/micro/go-micro/tunnel" @@ -18,12 +19,22 @@ import ( ) var ( - // ControlChannel is the name of the tunnel channel for passing contron message + // NetworkChannel is the name of the tunnel channel for passing network messages + NetworkChannel = "network" + // ControlChannel is the name of the tunnel channel for passing control message ControlChannel = "control" // DefaultLink is default network link DefaultLink = "network" ) +// node is network node +type node struct { + // Id is node id + Id string + // Address is node address + Address string +} + // network implements Network interface type network struct { // options configure the network @@ -39,11 +50,18 @@ type network struct { // client is network client client client.Client + // ctrlClient is ControlChannel client + ctrlClient transport.Client + // netClient is NetwrokChannel client + netClient transport.Client + sync.RWMutex // connected marks the network as connected connected bool // closed closes the network closed chan bool + // neighbours maps the node neighbourhood + neighbours map[node]map[node]bool } // newNetwork returns a new network node @@ -88,12 +106,13 @@ func newNetwork(opts ...Option) Network { ) return &network{ - options: options, - Router: options.Router, - Proxy: options.Proxy, - Tunnel: options.Tunnel, - server: server, - client: client, + options: options, + Router: options.Router, + Proxy: options.Proxy, + Tunnel: options.Tunnel, + server: server, + client: client, + neighbours: make(map[node]map[node]bool), } } @@ -107,6 +126,7 @@ func (n *network) Address() string { return n.Tunnel.Address() } +// resolveNodes resolves network nodes to addresses func (n *network) resolveNodes() ([]string, error) { // resolve the network address to network nodes records, err := n.options.Resolver.Resolve(n.options.Name) @@ -123,6 +143,7 @@ func (n *network) resolveNodes() ([]string, error) { return nodes, nil } +// resolve continuously resolves network nodes and initializes network tunnel with resolved addresses func (n *network) resolve() { resolve := time.NewTicker(ResolveTime) defer resolve.Stop() @@ -145,7 +166,180 @@ func (n *network) resolve() { } } -func (n *network) handleConn(conn tunnel.Conn, msg chan *transport.Message) { +// handleNetConn handles network announcement messages +func (n *network) handleNetConn(conn tunnel.Conn, msg chan *transport.Message) { + for { + m := new(transport.Message) + if err := conn.Recv(m); err != nil { + // TODO: should we bail here? + log.Debugf("Network tunnel [%s] receive error: %v", NetworkChannel, err) + return + } + + select { + case msg <- m: + case <-n.closed: + return + } + } +} + +// acceptNetConn accepts connections from NetworkChannel +func (n *network) acceptNetConn(l tunnel.Listener, recv chan *transport.Message) { + for { + // accept a connection + conn, err := l.Accept() + if err != nil { + // TODO: handle this + log.Debugf("Network tunnel [%s] accept error: %v", NetworkChannel, err) + return + } + + select { + case <-n.closed: + return + default: + // go handle NetworkChannel connection + go n.handleNetConn(conn, recv) + } + } +} + +// processNetChan processes messages received on NetworkChannel +func (n *network) processNetChan(l tunnel.Listener) { + // receive network message queue + recv := make(chan *transport.Message, 128) + + // accept NetworkChannel connections + go n.acceptNetConn(l, recv) + + for { + select { + case m := <-recv: + // switch on type of message and take action + switch m.Header["Micro-Method"] { + case "connect": + pbNetConnect := &pbNet.Connect{} + if err := proto.Unmarshal(m.Body, pbNetConnect); err != nil { + log.Debugf("Network tunnel [%s] connect unmarshal error: %v", NetworkChannel, err) + continue + } + // don't process your own messages + if pbNetConnect.Node.Id == n.options.Id { + continue + } + neighbour := node{ + Id: pbNetConnect.Node.Id, + Address: pbNetConnect.Node.Address, + } + n.Lock() + n.neighbours[neighbour] = make(map[node]bool) + n.Unlock() + case "neighbour": + pbNetNeighbour := &pbNet.Neighbour{} + if err := proto.Unmarshal(m.Body, pbNetNeighbour); err != nil { + log.Debugf("Network tunnel [%s] neighbour unmarshal error: %v", NetworkChannel, err) + continue + } + // don't process your own messages + if pbNetNeighbour.Node.Id == n.options.Id { + continue + } + neighbour := node{ + Id: pbNetNeighbour.Node.Id, + Address: pbNetNeighbour.Node.Address, + } + n.Lock() + // we override the existing neighbour map + n.neighbours[neighbour] = make(map[node]bool) + // store the neighbouring node and its neighbours + for _, pbNeighbour := range pbNetNeighbour.Neighbours { + neighbourNode := node{ + Id: pbNeighbour.Id, + Address: pbNeighbour.Address, + } + n.neighbours[neighbour][neighbourNode] = true + } + n.Unlock() + case "close": + pbNetClose := &pbNet.Close{} + if err := proto.Unmarshal(m.Body, pbNetClose); err != nil { + log.Debugf("Network tunnel [%s] close unmarshal error: %v", NetworkChannel, err) + continue + } + // don't process your own messages + if pbNetClose.Node.Id == n.options.Id { + continue + } + node := node{ + Id: pbNetClose.Node.Id, + Address: pbNetClose.Node.Address, + } + n.Lock() + delete(n.neighbours, node) + n.Unlock() + } + case <-n.closed: + return + } + } +} + +// announce announces node neighbourhood to the network +func (n *network) announce(client transport.Client) { + announce := time.NewTicker(AnnounceTime) + defer announce.Stop() + + for { + select { + case <-n.closed: + return + case <-announce.C: + n.RLock() + nodes := make([]*pbNet.Node, len(n.neighbours)) + i := 0 + for node, _ := range n.neighbours { + pbNode := &pbNet.Node{ + Id: node.Id, + Address: node.Address, + } + nodes[i] = pbNode + } + n.RUnlock() + + node := &pbNet.Node{ + Id: n.options.Id, + Address: n.options.Address, + } + pbNetNeighbour := &pbNet.Neighbour{ + Node: node, + Neighbours: nodes, + } + + body, err := proto.Marshal(pbNetNeighbour) + if err != nil { + // TODO: should we bail here? + log.Debugf("Network failed to marshal neighbour message: %v", err) + continue + } + // create transport message and chuck it down the pipe + m := transport.Message{ + Header: map[string]string{ + "Micro-Method": "neighbour", + }, + Body: body, + } + + if err := client.Send(&m); err != nil { + log.Debugf("Network failed to send neighbour messsage: %v", err) + continue + } + } + } +} + +// handleCtrlConn handles ControlChannel connections +func (n *network) handleCtrlConn(conn tunnel.Conn, msg chan *transport.Message) { for { m := new(transport.Message) if err := conn.Recv(m); err != nil { @@ -162,19 +356,34 @@ func (n *network) handleConn(conn tunnel.Conn, msg chan *transport.Message) { } } -func (n *network) process(l tunnel.Listener) { +// acceptCtrlConn accepts connections from ControlChannel +func (n *network) acceptCtrlConn(l tunnel.Listener, recv chan *transport.Message) { + for { + // accept a connection + conn, err := l.Accept() + if err != nil { + // TODO: handle this + log.Debugf("Network tunnel [%s] accept error: %v", ControlChannel, err) + return + } + + select { + case <-n.closed: + return + default: + // go handle ControlChannel connection + go n.handleCtrlConn(conn, recv) + } + } +} + +// process processes network advertisements +func (n *network) processCtrlChan(l tunnel.Listener) { // receive control message queue recv := make(chan *transport.Message, 128) - // accept a connection - conn, err := l.Accept() - if err != nil { - // TODO: handle this - log.Debugf("Network tunnel accept error: %v", err) - return - } - - go n.handleConn(conn, recv) + // accept ControlChannel cconnections + go n.acceptCtrlConn(l, recv) for { select { @@ -182,13 +391,13 @@ func (n *network) process(l tunnel.Listener) { // switch on type of message and take action switch m.Header["Micro-Method"] { case "advert": - pbAdvert := &pb.Advert{} - if err := proto.Unmarshal(m.Body, pbAdvert); err != nil { + pbRtrAdvert := &pbRtr.Advert{} + if err := proto.Unmarshal(m.Body, pbRtrAdvert); err != nil { continue } var events []*router.Event - for _, event := range pbAdvert.Events { + for _, event := range pbRtrAdvert.Events { route := router.Route{ Service: event.Route.Service, Address: event.Route.Address, @@ -200,16 +409,16 @@ func (n *network) process(l tunnel.Listener) { } e := &router.Event{ Type: router.EventType(event.Type), - Timestamp: time.Unix(0, pbAdvert.Timestamp), + Timestamp: time.Unix(0, pbRtrAdvert.Timestamp), Route: route, } events = append(events, e) } advert := &router.Advert{ - Id: pbAdvert.Id, - Type: router.AdvertType(pbAdvert.Type), - Timestamp: time.Unix(0, pbAdvert.Timestamp), - TTL: time.Duration(pbAdvert.Ttl), + Id: pbRtrAdvert.Id, + Type: router.AdvertType(pbRtrAdvert.Type), + Timestamp: time.Unix(0, pbRtrAdvert.Timestamp), + TTL: time.Duration(pbRtrAdvert.Ttl), Events: events, } @@ -231,10 +440,10 @@ func (n *network) advertise(client transport.Client, advertChan <-chan *router.A // process local adverts and randomly fire them at other nodes case advert := <-advertChan: // create a proto advert - var events []*pb.Event + var events []*pbRtr.Event for _, event := range advert.Events { // NOTE: we override the Gateway and Link fields here - route := &pb.Route{ + route := &pbRtr.Route{ Service: event.Route.Service, Address: event.Route.Address, Gateway: n.options.Address, @@ -243,23 +452,23 @@ func (n *network) advertise(client transport.Client, advertChan <-chan *router.A Link: DefaultLink, Metric: int64(event.Route.Metric), } - e := &pb.Event{ - Type: pb.EventType(event.Type), + e := &pbRtr.Event{ + Type: pbRtr.EventType(event.Type), Timestamp: event.Timestamp.UnixNano(), Route: route, } events = append(events, e) } - pbAdvert := &pb.Advert{ + pbRtrAdvert := &pbRtr.Advert{ Id: advert.Id, - Type: pb.AdvertType(advert.Type), + Type: pbRtr.AdvertType(advert.Type), Timestamp: advert.Timestamp.UnixNano(), Events: events, } - body, err := proto.Marshal(pbAdvert) + body, err := proto.Marshal(pbRtrAdvert) if err != nil { // TODO: should we bail here? - log.Debugf("Network failed to marshal message: %v", err) + log.Debugf("Network failed to marshal advert message: %v", err) continue } // create transport message and chuck it down the pipe @@ -271,7 +480,7 @@ func (n *network) advertise(client transport.Client, advertChan <-chan *router.A } if err := client.Send(&m); err != nil { - log.Debugf("Network failed to send advert %s: %v", pbAdvert.Id, err) + log.Debugf("Network failed to send advert %s: %v", pbRtrAdvert.Id, err) continue } case <-n.closed: @@ -307,13 +516,29 @@ func (n *network) Connect() error { ) // dial into ControlChannel to send route adverts - client, err := n.Tunnel.Dial(ControlChannel) + ctrlClient, err := n.Tunnel.Dial(ControlChannel) if err != nil { return err } + n.ctrlClient = ctrlClient + // listen on ControlChannel - listener, err := n.Tunnel.Listen(ControlChannel) + ctrlListener, err := n.Tunnel.Listen(ControlChannel) + if err != nil { + return err + } + + // dial into NetworkChannel to send network messages + netClient, err := n.Tunnel.Dial(NetworkChannel) + if err != nil { + return err + } + + n.netClient = netClient + + // listen on NetworkChannel + netListener, err := n.Tunnel.Listen(NetworkChannel) if err != nil { return err } @@ -321,9 +546,6 @@ func (n *network) Connect() error { // create closed channel n.closed = make(chan bool) - // keep resolving network nodes - go n.resolve() - // start the router if err := n.options.Router.Start(); err != nil { return err @@ -335,19 +557,48 @@ func (n *network) Connect() error { return err } - // advertise routes - go n.advertise(client, advertChan) - // accept and process routes - go n.process(listener) - // start the server if err := n.server.Start(); err != nil { return err } + // go resolving network nodes + go n.resolve() + // broadcast neighbourhood + go n.announce(netClient) + // listen to network messages + go n.processNetChan(netListener) + // advertise service routes + go n.advertise(ctrlClient, advertChan) + // accept and process routes + go n.processCtrlChan(ctrlListener) + // set connected to true n.connected = true + // send connect message to NetworkChannel + node := &pbNet.Node{ + Id: n.options.Id, + Address: n.options.Address, + } + pbNetConnect := &pbNet.Connect{ + Node: node, + } + + // only proceed with sending to NetworkChannel if marshal succeeds + if body, err := proto.Marshal(pbNetConnect); err == nil { + m := transport.Message{ + Header: map[string]string{ + "Micro-Method": "connect", + }, + Body: body, + } + + if err := netClient.Send(&m); err != nil { + log.Debugf("Network failed to send connect messsage: %v", err) + } + } + return nil } @@ -383,11 +634,39 @@ func (n *network) Close() error { case <-n.closed: return nil default: + // TODO: send close message to the network channel close(n.closed) // set connected to false n.connected = false } + // send close message only if we managed to connect to NetworkChannel + if n.netClient != nil { + // send connect message to NetworkChannel + node := &pbNet.Node{ + Id: n.options.Id, + Address: n.options.Address, + } + pbNetClose := &pbNet.Close{ + Node: node, + } + + // only proceed with sending to NetworkChannel if marshal succeeds + if body, err := proto.Marshal(pbNetClose); err == nil { + // create transport message and chuck it down the pipe + m := transport.Message{ + Header: map[string]string{ + "Micro-Method": "close", + }, + Body: body, + } + + if err := n.netClient.Send(&m); err != nil { + log.Debugf("Network failed to send close messsage: %v", err) + } + } + } + return n.close() } diff --git a/network/network.go b/network/network.go index 46801e0c..d9d10a55 100644 --- a/network/network.go +++ b/network/network.go @@ -13,8 +13,10 @@ var ( DefaultName = "go.micro" // DefaultAddress is default network address DefaultAddress = ":0" - // ResolveTime ddefines the time to periodically resolve network nodes + // ResolveTime defines time interval to periodically resolve network nodes ResolveTime = 1 * time.Minute + // AnnounceTime defines time interval to periodically announce node neighbours + AnnounceTime = 30 * time.Second ) // Network is micro network diff --git a/network/proto/network.micro.go b/network/proto/network.micro.go new file mode 100644 index 00000000..814e6dba --- /dev/null +++ b/network/proto/network.micro.go @@ -0,0 +1,21 @@ +// Code generated by protoc-gen-micro. DO NOT EDIT. +// source: network.proto + +package go_micro_network + +import ( + fmt "fmt" + proto "github.com/golang/protobuf/proto" + math "math" +) + +// Reference imports to suppress errors if they are not otherwise used. +var _ = proto.Marshal +var _ = fmt.Errorf +var _ = math.Inf + +// This is a compile-time assertion to ensure that this generated file +// is compatible with the proto package it is being compiled against. +// A compilation error at this line likely means your copy of the +// proto package needs to be updated. +const _ = proto.ProtoPackageIsVersion3 // please upgrade the proto package diff --git a/network/proto/network.pb.go b/network/proto/network.pb.go new file mode 100644 index 00000000..1a1d5948 --- /dev/null +++ b/network/proto/network.pb.go @@ -0,0 +1,227 @@ +// Code generated by protoc-gen-go. DO NOT EDIT. +// source: network.proto + +package go_micro_network + +import ( + fmt "fmt" + proto "github.com/golang/protobuf/proto" + math "math" +) + +// Reference imports to suppress errors if they are not otherwise used. +var _ = proto.Marshal +var _ = fmt.Errorf +var _ = math.Inf + +// This is a compile-time assertion to ensure that this generated file +// is compatible with the proto package it is being compiled against. +// A compilation error at this line likely means your copy of the +// proto package needs to be updated. +const _ = proto.ProtoPackageIsVersion3 // please upgrade the proto package + +// Node is network node +type Node struct { + // node ide + Id string `protobuf:"bytes,1,opt,name=id,proto3" json:"id,omitempty"` + // node address + Address string `protobuf:"bytes,2,opt,name=address,proto3" json:"address,omitempty"` + XXX_NoUnkeyedLiteral struct{} `json:"-"` + XXX_unrecognized []byte `json:"-"` + XXX_sizecache int32 `json:"-"` +} + +func (m *Node) Reset() { *m = Node{} } +func (m *Node) String() string { return proto.CompactTextString(m) } +func (*Node) ProtoMessage() {} +func (*Node) Descriptor() ([]byte, []int) { + return fileDescriptor_8571034d60397816, []int{0} +} + +func (m *Node) XXX_Unmarshal(b []byte) error { + return xxx_messageInfo_Node.Unmarshal(m, b) +} +func (m *Node) XXX_Marshal(b []byte, deterministic bool) ([]byte, error) { + return xxx_messageInfo_Node.Marshal(b, m, deterministic) +} +func (m *Node) XXX_Merge(src proto.Message) { + xxx_messageInfo_Node.Merge(m, src) +} +func (m *Node) XXX_Size() int { + return xxx_messageInfo_Node.Size(m) +} +func (m *Node) XXX_DiscardUnknown() { + xxx_messageInfo_Node.DiscardUnknown(m) +} + +var xxx_messageInfo_Node proto.InternalMessageInfo + +func (m *Node) GetId() string { + if m != nil { + return m.Id + } + return "" +} + +func (m *Node) GetAddress() string { + if m != nil { + return m.Address + } + return "" +} + +// Connect is sent when the node connects to the network +type Connect struct { + // network mode + Node *Node `protobuf:"bytes,1,opt,name=node,proto3" json:"node,omitempty"` + XXX_NoUnkeyedLiteral struct{} `json:"-"` + XXX_unrecognized []byte `json:"-"` + XXX_sizecache int32 `json:"-"` +} + +func (m *Connect) Reset() { *m = Connect{} } +func (m *Connect) String() string { return proto.CompactTextString(m) } +func (*Connect) ProtoMessage() {} +func (*Connect) Descriptor() ([]byte, []int) { + return fileDescriptor_8571034d60397816, []int{1} +} + +func (m *Connect) XXX_Unmarshal(b []byte) error { + return xxx_messageInfo_Connect.Unmarshal(m, b) +} +func (m *Connect) XXX_Marshal(b []byte, deterministic bool) ([]byte, error) { + return xxx_messageInfo_Connect.Marshal(b, m, deterministic) +} +func (m *Connect) XXX_Merge(src proto.Message) { + xxx_messageInfo_Connect.Merge(m, src) +} +func (m *Connect) XXX_Size() int { + return xxx_messageInfo_Connect.Size(m) +} +func (m *Connect) XXX_DiscardUnknown() { + xxx_messageInfo_Connect.DiscardUnknown(m) +} + +var xxx_messageInfo_Connect proto.InternalMessageInfo + +func (m *Connect) GetNode() *Node { + if m != nil { + return m.Node + } + return nil +} + +// Close is sent when the node disconnects from the network +type Close struct { + // network mode + Node *Node `protobuf:"bytes,1,opt,name=node,proto3" json:"node,omitempty"` + XXX_NoUnkeyedLiteral struct{} `json:"-"` + XXX_unrecognized []byte `json:"-"` + XXX_sizecache int32 `json:"-"` +} + +func (m *Close) Reset() { *m = Close{} } +func (m *Close) String() string { return proto.CompactTextString(m) } +func (*Close) ProtoMessage() {} +func (*Close) Descriptor() ([]byte, []int) { + return fileDescriptor_8571034d60397816, []int{2} +} + +func (m *Close) XXX_Unmarshal(b []byte) error { + return xxx_messageInfo_Close.Unmarshal(m, b) +} +func (m *Close) XXX_Marshal(b []byte, deterministic bool) ([]byte, error) { + return xxx_messageInfo_Close.Marshal(b, m, deterministic) +} +func (m *Close) XXX_Merge(src proto.Message) { + xxx_messageInfo_Close.Merge(m, src) +} +func (m *Close) XXX_Size() int { + return xxx_messageInfo_Close.Size(m) +} +func (m *Close) XXX_DiscardUnknown() { + xxx_messageInfo_Close.DiscardUnknown(m) +} + +var xxx_messageInfo_Close proto.InternalMessageInfo + +func (m *Close) GetNode() *Node { + if m != nil { + return m.Node + } + return nil +} + +// Neighbour is used to nnounce node neighbourhood +type Neighbour struct { + // network mode + Node *Node `protobuf:"bytes,1,opt,name=node,proto3" json:"node,omitempty"` + // neighbours + Neighbours []*Node `protobuf:"bytes,3,rep,name=neighbours,proto3" json:"neighbours,omitempty"` + XXX_NoUnkeyedLiteral struct{} `json:"-"` + XXX_unrecognized []byte `json:"-"` + XXX_sizecache int32 `json:"-"` +} + +func (m *Neighbour) Reset() { *m = Neighbour{} } +func (m *Neighbour) String() string { return proto.CompactTextString(m) } +func (*Neighbour) ProtoMessage() {} +func (*Neighbour) Descriptor() ([]byte, []int) { + return fileDescriptor_8571034d60397816, []int{3} +} + +func (m *Neighbour) XXX_Unmarshal(b []byte) error { + return xxx_messageInfo_Neighbour.Unmarshal(m, b) +} +func (m *Neighbour) XXX_Marshal(b []byte, deterministic bool) ([]byte, error) { + return xxx_messageInfo_Neighbour.Marshal(b, m, deterministic) +} +func (m *Neighbour) XXX_Merge(src proto.Message) { + xxx_messageInfo_Neighbour.Merge(m, src) +} +func (m *Neighbour) XXX_Size() int { + return xxx_messageInfo_Neighbour.Size(m) +} +func (m *Neighbour) XXX_DiscardUnknown() { + xxx_messageInfo_Neighbour.DiscardUnknown(m) +} + +var xxx_messageInfo_Neighbour proto.InternalMessageInfo + +func (m *Neighbour) GetNode() *Node { + if m != nil { + return m.Node + } + return nil +} + +func (m *Neighbour) GetNeighbours() []*Node { + if m != nil { + return m.Neighbours + } + return nil +} + +func init() { + proto.RegisterType((*Node)(nil), "go.micro.network.Node") + proto.RegisterType((*Connect)(nil), "go.micro.network.Connect") + proto.RegisterType((*Close)(nil), "go.micro.network.Close") + proto.RegisterType((*Neighbour)(nil), "go.micro.network.Neighbour") +} + +func init() { proto.RegisterFile("network.proto", fileDescriptor_8571034d60397816) } + +var fileDescriptor_8571034d60397816 = []byte{ + // 173 bytes of a gzipped FileDescriptorProto + 0x1f, 0x8b, 0x08, 0x00, 0x00, 0x00, 0x00, 0x00, 0x02, 0xff, 0xe2, 0xe2, 0xcd, 0x4b, 0x2d, 0x29, + 0xcf, 0x2f, 0xca, 0xd6, 0x2b, 0x28, 0xca, 0x2f, 0xc9, 0x17, 0x12, 0x48, 0xcf, 0xd7, 0xcb, 0xcd, + 0x4c, 0x2e, 0xca, 0xd7, 0x83, 0x8a, 0x2b, 0x19, 0x70, 0xb1, 0xf8, 0xe5, 0xa7, 0xa4, 0x0a, 0xf1, + 0x71, 0x31, 0x65, 0xa6, 0x48, 0x30, 0x2a, 0x30, 0x6a, 0x70, 0x06, 0x31, 0x65, 0xa6, 0x08, 0x49, + 0x70, 0xb1, 0x27, 0xa6, 0xa4, 0x14, 0xa5, 0x16, 0x17, 0x4b, 0x30, 0x81, 0x05, 0x61, 0x5c, 0x25, + 0x53, 0x2e, 0x76, 0xe7, 0xfc, 0xbc, 0xbc, 0xd4, 0xe4, 0x12, 0x21, 0x2d, 0x2e, 0x96, 0xbc, 0xfc, + 0x94, 0x54, 0xb0, 0x36, 0x6e, 0x23, 0x31, 0x3d, 0x74, 0xd3, 0xf5, 0x40, 0x46, 0x07, 0x81, 0xd5, + 0x28, 0x19, 0x73, 0xb1, 0x3a, 0xe7, 0xe4, 0x17, 0xa7, 0x92, 0xa4, 0x29, 0x9f, 0x8b, 0xd3, 0x2f, + 0x35, 0x33, 0x3d, 0x23, 0x29, 0xbf, 0xb4, 0x88, 0x14, 0x8d, 0x42, 0x66, 0x5c, 0x5c, 0x79, 0x30, + 0x8d, 0xc5, 0x12, 0xcc, 0x0a, 0xcc, 0x78, 0x74, 0x20, 0xa9, 0x4c, 0x62, 0x03, 0x87, 0x93, 0x31, + 0x20, 0x00, 0x00, 0xff, 0xff, 0xb8, 0x74, 0x00, 0x71, 0x38, 0x01, 0x00, 0x00, +} diff --git a/network/proto/network.proto b/network/proto/network.proto new file mode 100644 index 00000000..10483116 --- /dev/null +++ b/network/proto/network.proto @@ -0,0 +1,31 @@ +syntax = "proto3"; + +package go.micro.network; + +// Node is network node +message Node { + // node ide + string id = 1; + // node address + string address = 2; +} + +// Connect is sent when the node connects to the network +message Connect { + // network mode + Node node = 1; +} + +// Close is sent when the node disconnects from the network +message Close { + // network mode + Node node = 1; +} + +// Neighbour is used to nnounce node neighbourhood +message Neighbour { + // network mode + Node node = 1; + // neighbours + repeated Node neighbours = 3; +}