Commit a4bfa5bc authored by suyanlong's avatar suyanlong

Add HandleIBTPX and getRemotePeerInfo function

parent a2e0aa28
Pipeline #7952 canceled with stages
......@@ -3,13 +3,13 @@ package appchain
import (
"context"
"fmt"
"github.com/link33/sidecar/internal/port"
"strconv"
"strings"
"time"
"github.com/Rican7/retry"
"github.com/Rican7/retry/strategy"
"github.com/link33/sidecar/internal/port"
"github.com/link33/sidecar/internal/txcrypto"
"github.com/link33/sidecar/model/pb"
"github.com/link33/sidecar/pkg/plugins"
......
......@@ -32,8 +32,6 @@ func (swarm *Swarm) RegisterMsgHandler(messageType pb.Message_Type, handler Mess
func (swarm *Swarm) RegisterConnectHandler(handler ConnectHandler) error {
swarm.lock.Lock()
defer swarm.lock.Unlock()
swarm.connectHandlers = append(swarm.connectHandlers, handler)
return nil
}
......@@ -6,6 +6,7 @@ import (
"github.com/link33/sidecar/internal/router"
"github.com/link33/sidecar/model/pb"
"github.com/meshplus/bitxhub-kit/crypto"
"github.com/sirupsen/logrus"
)
type local struct {
......@@ -14,6 +15,7 @@ type local struct {
tag string
rev chan *pb.Message
rout router.Router
logger logrus.FieldLogger
}
func (l *local) ID() string {
......@@ -58,3 +60,15 @@ func (l *local) ListenIBTPX() <-chan *pb.Message {
}
// sidecar 节点
func (l *local) HandleGetPeerInfoMessage(p port.Port, message *pb.Message) {
data := &pb.PeerInfo{
ID: l.ID(),
Tag: l.Tag(),
}
msg, _ := data.Marshal()
retMsg := Message(pb.Message_ACK, true, msg)
err := p.AsyncSend(retMsg)
if err != nil {
l.logger.Error(err)
}
}
......@@ -107,6 +107,19 @@ func (swarm *Swarm) Start() error {
return fmt.Errorf("register get address msg handler: %w", err)
}
if err := swarm.RegisterMsgHandler(pb.Message_PEER_INFO_GET, l.HandleGetPeerInfoMessage); err != nil {
return fmt.Errorf("register get peer info msg handler: %w", err)
}
if err := swarm.RegisterMultiMsgHandler([]pb.Message_Type{
pb.Message_IBTP_SEND,
pb.Message_IBTP_GET,
pb.Message_IBTP_RECEIPT_SEND,
pb.Message_IBTP_RECEIPT_GET,
}, swarm.HandleIBTPX); err != nil {
return fmt.Errorf("register handle IBTPX msg handler: %w", err)
}
if err := swarm.p2p.Start(); err != nil {
return fmt.Errorf("p2p module start: %w", err)
}
......@@ -181,30 +194,51 @@ func (swarm *Swarm) handleMessage(s network.Stream, data []byte) {
return
}
t := m.Type
switch {
// 接收其它sidecar节点发过来的交易、请求等。主要是IBTP结构相关数据。
case t == pb.Message_IBTP_SEND || t == pb.Message_IBTP_GET || t == pb.Message_IBTP_RECEIPT_SEND || t == pb.Message_IBTP_RECEIPT_GET:
p, is := swarm.router.Load(s.RemotePeerID())
if is {
ps, iss := p.(*sidecar)
if iss {
ps.rev <- m
handler, ok := swarm.msgHandlers.Load(m.Type)
if !ok {
swarm.logger.WithFields(logrus.Fields{
"error": fmt.Errorf("can't handle msg[type: %v]", m.Type),
"type": m.Type.String(),
}).Error("Handle message")
return
}
msgHandler, ok := handler.(MessageHandler)
if !ok {
swarm.logger.WithFields(logrus.Fields{
"error": fmt.Errorf("invalid handler for msg [type: %v]", m.Type),
"type": m.Type.String(),
}).Error("Handle message")
return
}
addr, _ := peer.AddrInfoFromP2pAddr(s.RemotePeerAddr())
rec := make(chan *pb.Message)
newPort := &sidecar{
id: addr.ID.String(),
p := &sidecar{
id: s.RemotePeerID(),
swarm: swarm,
tag: "",
rev: rec,
}
swarm.router.Add(newPort)
rec <- m
default: //非IBTP结构相关数据,转发给local处理
//TODO
msgHandler(p, m)
}
// 接收其它sidecar节点发过来的交易、请求等。主要是IBTP结构相关数据。
func (swarm *Swarm) HandleIBTPX(pt port.Port, m *pb.Message) {
p, is := swarm.router.Load(pt.ID())
if is {
ps, iss := p.(*sidecar)
if iss {
ps.rev <- m
}
} else {
ppt := pt.(*sidecar)
info, err := swarm.getRemotePeerInfo(ppt.ID())
if err != nil {
swarm.logger.Error(err)
}
ppt.id = info.GetID()
_ = swarm.router.Add(ppt)
ppt.rev <- m
}
}
......@@ -361,6 +395,25 @@ func (swarm *Swarm) getRemoteAddress(id peer.ID) (string, error) {
return string(ret.Payload.Data), nil
}
func (swarm *Swarm) getRemotePeerInfo(id string) (*pb.PeerInfo, error) {
msg := Message(pb.Message_PEER_INFO_GET, true, nil)
reqData, err := msg.Marshal()
if err != nil {
return nil, err
}
retData, err := swarm.p2p.Send(id.String(), reqData) //同步获取数据
if err != nil {
return nil, fmt.Errorf("sync send: %w", err)
}
ret := &pb.Message{}
if err := ret.Unmarshal(retData); err != nil {
return nil, err
}
info := &pb.PeerInfo{}
err = info.Unmarshal(ret.Payload.Data)
return info, err
}
func (swarm *Swarm) FindProviders(id string) (string, error) {
format := cid.V0Builder{}
toCid, err := format.Sum([]byte(id))
......
......@@ -151,7 +151,7 @@ func (r *router) Route(msg *pb.Message) error {
return nil
}
func (r *router) firstRoute(ibtp *pb.Message) {
func (r *router) firstRoute(msg *pb.Message) {
panic("implement me")
}
......
......@@ -37,6 +37,7 @@ const (
Message_IBTP_SEND Message_Type = 8
Message_IBTP_RECEIPT_SEND Message_Type = 9
Message_IBTP_RECEIPT_GET Message_Type = 10
Message_PEER_INFO_GET Message_Type = 11
)
var Message_Type_name = map[int32]string{
......@@ -51,6 +52,7 @@ var Message_Type_name = map[int32]string{
8: "IBTP_SEND",
9: "IBTP_RECEIPT_SEND",
10: "IBTP_RECEIPT_GET",
11: "PEER_INFO_GET",
}
var Message_Type_value = map[string]int32{
......@@ -65,6 +67,7 @@ var Message_Type_value = map[string]int32{
"IBTP_SEND": 8,
"IBTP_RECEIPT_SEND": 9,
"IBTP_RECEIPT_GET": 10,
"PEER_INFO_GET": 11,
}
func (x Message_Type) String() string {
......@@ -187,38 +190,93 @@ func (m *Pack) GetData() []byte {
return nil
}
type PeerInfo struct {
ID string `protobuf:"bytes,1,opt,name=ID,proto3" json:"ID,omitempty"`
Tag string `protobuf:"bytes,2,opt,name=Tag,proto3" json:"Tag,omitempty"`
}
func (m *PeerInfo) Reset() { *m = PeerInfo{} }
func (m *PeerInfo) String() string { return proto.CompactTextString(m) }
func (*PeerInfo) ProtoMessage() {}
func (*PeerInfo) Descriptor() ([]byte, []int) {
return fileDescriptor_33c57e4bae7b9afd, []int{2}
}
func (m *PeerInfo) XXX_Unmarshal(b []byte) error {
return m.Unmarshal(b)
}
func (m *PeerInfo) XXX_Marshal(b []byte, deterministic bool) ([]byte, error) {
if deterministic {
return xxx_messageInfo_PeerInfo.Marshal(b, m, deterministic)
} else {
b = b[:cap(b)]
n, err := m.MarshalToSizedBuffer(b)
if err != nil {
return nil, err
}
return b[:n], nil
}
}
func (m *PeerInfo) XXX_Merge(src proto.Message) {
xxx_messageInfo_PeerInfo.Merge(m, src)
}
func (m *PeerInfo) XXX_Size() int {
return m.Size()
}
func (m *PeerInfo) XXX_DiscardUnknown() {
xxx_messageInfo_PeerInfo.DiscardUnknown(m)
}
var xxx_messageInfo_PeerInfo proto.InternalMessageInfo
func (m *PeerInfo) GetID() string {
if m != nil {
return m.ID
}
return ""
}
func (m *PeerInfo) GetTag() string {
if m != nil {
return m.Tag
}
return ""
}
func init() {
proto.RegisterEnum("pb.Message_Type", Message_Type_name, Message_Type_value)
proto.RegisterType((*Message)(nil), "pb.Message")
proto.RegisterType((*Pack)(nil), "pb.Pack")
proto.RegisterType((*PeerInfo)(nil), "pb.PeerInfo")
}
func init() { proto.RegisterFile("message.proto", fileDescriptor_33c57e4bae7b9afd) }
var fileDescriptor_33c57e4bae7b9afd = []byte{
// 339 bytes of a gzipped FileDescriptorProto
0x1f, 0x8b, 0x08, 0x00, 0x00, 0x00, 0x00, 0x00, 0x02, 0xff, 0x54, 0x91, 0xcd, 0x4e, 0xea, 0x40,
0x14, 0xc7, 0x3b, 0xa5, 0x97, 0x96, 0xc3, 0xd7, 0xdc, 0xc3, 0xbd, 0xb1, 0xab, 0x86, 0x34, 0x2e,
0x88, 0x8b, 0x2e, 0xf0, 0x09, 0x0a, 0x9d, 0x60, 0x23, 0xe0, 0x64, 0x3a, 0x2c, 0x5c, 0x91, 0x22,
0x8d, 0x31, 0xa8, 0x6d, 0x80, 0x98, 0xf0, 0x16, 0x3e, 0x93, 0x2b, 0x97, 0xc4, 0x95, 0x4b, 0x03,
0x2f, 0x62, 0x66, 0x10, 0x12, 0x77, 0xf3, 0xff, 0xfd, 0xe6, 0xcc, 0xc7, 0x39, 0x50, 0x7f, 0xca,
0x56, 0xab, 0xf4, 0x3e, 0x0b, 0x8a, 0x65, 0xbe, 0xce, 0xd1, 0x2c, 0x66, 0xfe, 0x9b, 0x09, 0xf6,
0xe8, 0x40, 0xf1, 0x1c, 0xac, 0xf5, 0xa6, 0xc8, 0x5c, 0xd2, 0x26, 0x9d, 0x46, 0x97, 0x06, 0xc5,
0x2c, 0xf8, 0x51, 0x81, 0xdc, 0x14, 0x99, 0xd0, 0x16, 0x7d, 0xb0, 0x8b, 0x74, 0xf3, 0x98, 0xa7,
0x73, 0xd7, 0x6c, 0x93, 0x4e, 0xb5, 0xeb, 0xa8, 0x8d, 0x3c, 0xbd, 0x5b, 0x88, 0xa3, 0x40, 0x17,
0xec, 0x97, 0x6c, 0xb9, 0x7a, 0xc8, 0x9f, 0xdd, 0x52, 0x9b, 0x74, 0x2a, 0xe2, 0x18, 0xfd, 0x0f,
0x02, 0x96, 0x3a, 0x0c, 0x6d, 0x28, 0x85, 0xfd, 0x6b, 0x6a, 0x60, 0x13, 0xaa, 0x61, 0x14, 0x09,
0x96, 0x24, 0xd3, 0x01, 0x93, 0x94, 0xe0, 0x7f, 0xf8, 0x1b, 0x72, 0xde, 0xbf, 0x0a, 0xe3, 0xf1,
0x54, 0xb0, 0x41, 0x9c, 0x48, 0x26, 0xa8, 0x89, 0x2d, 0x68, 0x9e, 0xf0, 0x84, 0x47, 0xa1, 0x64,
0xb4, 0x84, 0x14, 0x6a, 0x27, 0xa8, 0xaa, 0x2d, 0x3c, 0x83, 0x56, 0x3c, 0x96, 0x4c, 0x1c, 0xd8,
0x88, 0xc9, 0x50, 0x8b, 0x3f, 0xea, 0x1e, 0x31, 0x19, 0xb2, 0x69, 0xc4, 0xf8, 0xf0, 0xe6, 0x96,
0x96, 0xb1, 0x06, 0x4e, 0xdc, 0x93, 0x5c, 0x6b, 0x1b, 0xeb, 0x50, 0xd1, 0x29, 0x61, 0xe3, 0x88,
0x3a, 0xea, 0x11, 0x3a, 0x0a, 0xd6, 0x67, 0x31, 0x97, 0x07, 0x5c, 0xc1, 0x7f, 0x40, 0x7f, 0x61,
0x55, 0x0b, 0xfe, 0x05, 0x58, 0xea, 0xff, 0xd8, 0x00, 0x33, 0x5f, 0xe8, 0xf6, 0x39, 0xc2, 0xcc,
0x17, 0x88, 0x60, 0xcd, 0xd3, 0x75, 0xaa, 0xfb, 0x54, 0x13, 0x7a, 0xdd, 0x73, 0xdf, 0x77, 0x1e,
0xd9, 0xee, 0x3c, 0xf2, 0xb5, 0xf3, 0xc8, 0xeb, 0xde, 0x33, 0xb6, 0x7b, 0xcf, 0xf8, 0xdc, 0x7b,
0xc6, 0xac, 0xac, 0xa7, 0x72, 0xf9, 0x1d, 0x00, 0x00, 0xff, 0xff, 0x8b, 0xda, 0x49, 0x68, 0xa6,
0x01, 0x00, 0x00,
// 381 bytes of a gzipped FileDescriptorProto
0x1f, 0x8b, 0x08, 0x00, 0x00, 0x00, 0x00, 0x00, 0x02, 0xff, 0x54, 0x91, 0xc1, 0x8e, 0xd3, 0x30,
0x14, 0x45, 0xe3, 0x24, 0x4c, 0x92, 0xd7, 0x76, 0xc6, 0xf3, 0x06, 0x44, 0x56, 0x51, 0x15, 0xb1,
0xa8, 0x10, 0xca, 0x62, 0xf8, 0x82, 0x4c, 0x63, 0x06, 0x8b, 0x36, 0xb5, 0x1c, 0x77, 0xc1, 0x2a,
0x4a, 0x69, 0xa8, 0x50, 0xa1, 0x89, 0xda, 0x0a, 0xa9, 0x7f, 0xc1, 0x67, 0xb1, 0xec, 0x06, 0x89,
0x25, 0x6a, 0xc5, 0x7f, 0x20, 0xbb, 0xb4, 0xd2, 0xec, 0xde, 0x3d, 0xf7, 0x3e, 0xdb, 0xba, 0x86,
0xde, 0xb7, 0x7a, 0xb3, 0xa9, 0x16, 0x75, 0xd2, 0xae, 0x9b, 0x6d, 0x83, 0x76, 0x3b, 0x8b, 0x7f,
0xd9, 0xe0, 0x8d, 0x4f, 0x14, 0x5f, 0x81, 0xbb, 0xdd, 0xb5, 0x75, 0x48, 0xfa, 0x64, 0x70, 0x7d,
0x4f, 0x93, 0x76, 0x96, 0xfc, 0xb7, 0x12, 0xb5, 0x6b, 0x6b, 0x69, 0x5c, 0x8c, 0xc1, 0x6b, 0xab,
0xdd, 0xd7, 0xa6, 0x9a, 0x87, 0x76, 0x9f, 0x0c, 0x3a, 0xf7, 0xbe, 0x0e, 0x8a, 0xea, 0xd3, 0x52,
0x9e, 0x0d, 0x0c, 0xc1, 0xfb, 0x5e, 0xaf, 0x37, 0x5f, 0x9a, 0x55, 0xe8, 0xf4, 0xc9, 0x20, 0x90,
0x67, 0x19, 0xff, 0x25, 0xe0, 0xea, 0xc3, 0xd0, 0x03, 0x27, 0x1d, 0x7e, 0xa0, 0x16, 0xde, 0x40,
0x27, 0xcd, 0x32, 0xc9, 0x8a, 0xa2, 0x7c, 0x64, 0x8a, 0x12, 0x7c, 0x01, 0xb7, 0xa9, 0x10, 0xc3,
0xf7, 0x29, 0xcf, 0x4b, 0xc9, 0x1e, 0x79, 0xa1, 0x98, 0xa4, 0x36, 0xde, 0xc1, 0xcd, 0x05, 0x4f,
0x45, 0x96, 0x2a, 0x46, 0x1d, 0xa4, 0xd0, 0xbd, 0x40, 0xbd, 0xed, 0xe2, 0x4b, 0xb8, 0xe3, 0xb9,
0x62, 0xf2, 0xc4, 0xc6, 0x4c, 0xa5, 0xc6, 0x78, 0xa6, 0xef, 0x91, 0xd3, 0x11, 0x2b, 0x33, 0x26,
0x46, 0x93, 0x8f, 0xf4, 0x0a, 0xbb, 0xe0, 0xf3, 0x07, 0x25, 0x8c, 0xed, 0x61, 0x0f, 0x02, 0xa3,
0x0a, 0x96, 0x67, 0xd4, 0xd7, 0x8f, 0x30, 0x52, 0xb2, 0x21, 0xe3, 0x42, 0x9d, 0x70, 0x80, 0xcf,
0x81, 0x3e, 0xc1, 0x7a, 0x17, 0xf0, 0x16, 0x7a, 0x82, 0x31, 0x59, 0xf2, 0xfc, 0xdd, 0xc4, 0xa0,
0x4e, 0xfc, 0x1a, 0x5c, 0x5d, 0x09, 0x5e, 0x83, 0xdd, 0x2c, 0x4d, 0xa3, 0xbe, 0xb4, 0x9b, 0x25,
0x22, 0xb8, 0xf3, 0x6a, 0x5b, 0x99, 0xea, 0xba, 0xd2, 0xcc, 0xf1, 0x1b, 0xf0, 0x45, 0x5d, 0xaf,
0xf9, 0xea, 0x73, 0xa3, 0xf3, 0x3c, 0x33, 0xf9, 0x40, 0xda, 0x3c, 0x43, 0x0a, 0x8e, 0xaa, 0x16,
0x26, 0x1e, 0x48, 0x3d, 0x3e, 0x84, 0x3f, 0x0f, 0x11, 0xd9, 0x1f, 0x22, 0xf2, 0xe7, 0x10, 0x91,
0x1f, 0xc7, 0xc8, 0xda, 0x1f, 0x23, 0xeb, 0xf7, 0x31, 0xb2, 0x66, 0x57, 0xe6, 0x5b, 0xdf, 0xfe,
0x0b, 0x00, 0x00, 0xff, 0xff, 0x37, 0x96, 0xad, 0xcc, 0xe7, 0x01, 0x00, 0x00,
}
func (m *Message) Marshal() (dAtA []byte, err error) {
......@@ -308,6 +366,43 @@ func (m *Pack) MarshalToSizedBuffer(dAtA []byte) (int, error) {
return len(dAtA) - i, nil
}
func (m *PeerInfo) Marshal() (dAtA []byte, err error) {
size := m.Size()
dAtA = make([]byte, size)
n, err := m.MarshalToSizedBuffer(dAtA[:size])
if err != nil {
return nil, err
}
return dAtA[:n], nil
}
func (m *PeerInfo) MarshalTo(dAtA []byte) (int, error) {
size := m.Size()
return m.MarshalToSizedBuffer(dAtA[:size])
}
func (m *PeerInfo) MarshalToSizedBuffer(dAtA []byte) (int, error) {
i := len(dAtA)
_ = i
var l int
_ = l
if len(m.Tag) > 0 {
i -= len(m.Tag)
copy(dAtA[i:], m.Tag)
i = encodeVarintMessage(dAtA, i, uint64(len(m.Tag)))
i--
dAtA[i] = 0x12
}
if len(m.ID) > 0 {
i -= len(m.ID)
copy(dAtA[i:], m.ID)
i = encodeVarintMessage(dAtA, i, uint64(len(m.ID)))
i--
dAtA[i] = 0xa
}
return len(dAtA) - i, nil
}
func encodeVarintMessage(dAtA []byte, offset int, v uint64) int {
offset -= sovMessage(v)
base := offset
......@@ -355,6 +450,23 @@ func (m *Pack) Size() (n int) {
return n
}
func (m *PeerInfo) Size() (n int) {
if m == nil {
return 0
}
var l int
_ = l
l = len(m.ID)
if l > 0 {
n += 1 + l + sovMessage(uint64(l))
}
l = len(m.Tag)
if l > 0 {
n += 1 + l + sovMessage(uint64(l))
}
return n
}
func sovMessage(x uint64) (n int) {
return (math_bits.Len64(x|1) + 6) / 7
}
......@@ -602,6 +714,120 @@ func (m *Pack) Unmarshal(dAtA []byte) error {
}
return nil
}
func (m *PeerInfo) Unmarshal(dAtA []byte) error {
l := len(dAtA)
iNdEx := 0
for iNdEx < l {
preIndex := iNdEx
var wire uint64
for shift := uint(0); ; shift += 7 {
if shift >= 64 {
return ErrIntOverflowMessage
}
if iNdEx >= l {
return io.ErrUnexpectedEOF
}
b := dAtA[iNdEx]
iNdEx++
wire |= uint64(b&0x7F) << shift
if b < 0x80 {
break
}
}
fieldNum := int32(wire >> 3)
wireType := int(wire & 0x7)
if wireType == 4 {
return fmt.Errorf("proto: PeerInfo: wiretype end group for non-group")
}
if fieldNum <= 0 {
return fmt.Errorf("proto: PeerInfo: illegal tag %d (wire type %d)", fieldNum, wire)
}
switch fieldNum {
case 1:
if wireType != 2 {
return fmt.Errorf("proto: wrong wireType = %d for field ID", wireType)
}
var stringLen uint64
for shift := uint(0); ; shift += 7 {
if shift >= 64 {
return ErrIntOverflowMessage
}
if iNdEx >= l {
return io.ErrUnexpectedEOF
}
b := dAtA[iNdEx]
iNdEx++
stringLen |= uint64(b&0x7F) << shift
if b < 0x80 {
break
}
}
intStringLen := int(stringLen)
if intStringLen < 0 {
return ErrInvalidLengthMessage
}
postIndex := iNdEx + intStringLen
if postIndex < 0 {
return ErrInvalidLengthMessage
}
if postIndex > l {
return io.ErrUnexpectedEOF
}
m.ID = string(dAtA[iNdEx:postIndex])
iNdEx = postIndex
case 2:
if wireType != 2 {
return fmt.Errorf("proto: wrong wireType = %d for field Tag", wireType)
}
var stringLen uint64
for shift := uint(0); ; shift += 7 {
if shift >= 64 {
return ErrIntOverflowMessage
}
if iNdEx >= l {
return io.ErrUnexpectedEOF
}
b := dAtA[iNdEx]
iNdEx++
stringLen |= uint64(b&0x7F) << shift
if b < 0x80 {
break
}
}
intStringLen := int(stringLen)
if intStringLen < 0 {
return ErrInvalidLengthMessage
}
postIndex := iNdEx + intStringLen
if postIndex < 0 {
return ErrInvalidLengthMessage
}
if postIndex > l {
return io.ErrUnexpectedEOF
}
m.Tag = string(dAtA[iNdEx:postIndex])
iNdEx = postIndex
default:
iNdEx = preIndex
skippy, err := skipMessage(dAtA[iNdEx:])
if err != nil {
return err
}
if (skippy < 0) || (iNdEx+skippy) < 0 {
return ErrInvalidLengthMessage
}
if (iNdEx + skippy) > l {
return io.ErrUnexpectedEOF
}
iNdEx += skippy
}
}
if iNdEx > l {
return io.ErrUnexpectedEOF
}
return nil
}
func skipMessage(dAtA []byte) (n int, err error) {
l := len(dAtA)
iNdEx := 0
......
......@@ -16,6 +16,7 @@ message Message {
IBTP_SEND = 8;
IBTP_RECEIPT_SEND = 9;
IBTP_RECEIPT_GET = 10;
PEER_INFO_GET = 11;
}
Type type = 1;
Pack payload = 2;
......@@ -26,3 +27,10 @@ message Pack {
bool ok = 1;
bytes data = 2;
}
message PeerInfo {
string ID = 1;
string Tag = 2;
}
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment