use protobuf while communicate between peers

network/p2p/p2p.go

@@ -15,18 +15,21 @@ import (
 	dutil "github.com/libp2p/go-libp2p/p2p/discovery/util"
 	"github.com/topology-gg/gram/config"
 	ex "github.com/topology-gg/gram/execution"
+	pbcodec "github.com/topology-gg/gram/proto"
+	"github.com/topology-gg/gram/proto/gen/gram/base"
 )
 
 type P2P struct {
-	ctx       context.Context
-	errCh     chan error
-	executor  ex.Execution
-	host      host.Host
-	namespace string
-	maxPeers  int
-	port      int
-	pubsub    *pubsub.PubSub
-	streams   []Stream
+	ctx        context.Context
+	errCh      chan error
+	executor   ex.Execution
+	host       host.Host
+	namespace  string
+	maxPeers   int
+	port       int
+	pubsub     *pubsub.PubSub
+	streams    []Stream
+	serializer pbcodec.Serializer
 }
 
 type Stream struct {
@@ -59,16 +62,19 @@ func NewP2P(ctx context.Context, errCh chan error, executor ex.Execution, cfg *c
 		streams[i] = Stream{name: name}
 	}
 
+	serializer := &pbcodec.ProtoBufSerializer{}
+
 	return &P2P{
-		ctx:       ctx,
-		errCh:     errCh,
-		executor:  executor,
-		host:      host,
-		namespace: namespace,
-		maxPeers:  maxPeers,
-		port:      port,
-		pubsub:    gossipsub,
-		streams:   streams,
+		ctx:        ctx,
+		errCh:      errCh,
+		executor:   executor,
+		host:       host,
+		namespace:  namespace,
+		maxPeers:   maxPeers,
+		port:       port,
+		pubsub:     gossipsub,
+		streams:    streams,
+		serializer: serializer,
 	}, nil
 }
 
@@ -78,8 +84,14 @@ func (p2p *P2P) Start() {
 }
 
 func (p2p *P2P) Publish(message string) {
+	msg, err := p2p.serializer.Marshal(&base.HelloRequest{Name: message})
+	if err != nil {
+		p2p.errCh <- err
+		return
+	}
+
 	for i := range p2p.streams {
-		if err := p2p.streams[i].topic.Publish(p2p.ctx, []byte(message)); err != nil {
+		if err := p2p.streams[i].topic.Publish(p2p.ctx, msg); err != nil {
 			fmt.Println("(Network) Failed to publish to topic:", p2p.streams[i].name, err)
 		}
 	}
@@ -211,6 +223,12 @@ func (p2p *P2P) p2pMessageHandler(subscription *pubsub.Subscription) {
 			continue
 		}
 
+		var msg base.HelloRequest
+		if err := p2p.serializer.Unmarshal(message.Data, &msg); err != nil {
+			p2p.errCh <- err
+			continue
+		}
+
 		p2p.executor.Execute(string(message.Message.Data))
 	}
 }

proto/interface.go

@@ -0,0 +1,6 @@
+package pbcodec
+
+type Serializer interface {
+	Marshal(data interface{}) ([]byte, error)
+	Unmarshal(data []byte, v interface{}) error
+}

proto/pbcodec.go

@@ -0,0 +1,13 @@
+package pbcodec
+
+import "google.golang.org/protobuf/proto"
+
+type ProtoBufSerializer struct{}
+
+func (p *ProtoBufSerializer) Marshal(data interface{}) ([]byte, error) {
+	return proto.Marshal(data.(proto.Message))
+}
+
+func (p *ProtoBufSerializer) Unmarshal(data []byte, v interface{}) error {
+	return proto.Unmarshal(data, v.(proto.Message))
+}