Add message service to handle streams

go.mod

@@ -10,9 +10,10 @@ require (
 	github.com/libp2p/go-libp2p v0.20.3
 	github.com/libp2p/go-libp2p-core v0.16.1
 	github.com/mr-tron/base58 v1.2.0
+	github.com/multiformats/go-multibase v0.0.3
 	github.com/multiformats/go-multihash v0.1.0
 	github.com/stretchr/testify v1.7.0
-	github.com/textileio/go-threads v1.0.2-0.20210304072541-d0f91da84404
+	github.com/cheggaaa/mb v1.0.3
 	go.uber.org/zap v1.21.0
 	gopkg.in/yaml.v3 v3.0.1
 	storj.io/drpc v0.0.32
@@ -24,13 +25,8 @@ require (
 	github.com/btcsuite/btcd/chaincfg/chainhash v1.0.1 // indirect
 	github.com/davecgh/go-spew v1.1.1 // indirect
 	github.com/decred/dcrd/dcrec/secp256k1/v4 v4.0.1 // indirect
-	github.com/dgrijalva/jwt-go v3.2.0+incompatible // indirect
 	github.com/fogleman/gg v1.3.0 // indirect
-	github.com/gogo/googleapis v1.3.1 // indirect
-	github.com/gogo/status v1.1.0 // indirect
 	github.com/golang/freetype v0.0.0-20170609003504-e2365dfdc4a0 // indirect
-	github.com/golang/protobuf v1.5.2 // indirect
-	github.com/grpc-ecosystem/go-grpc-middleware v1.2.1 // indirect
 	github.com/klauspost/cpuid/v2 v2.0.9 // indirect
 	github.com/libp2p/go-buffer-pool v0.0.2 // indirect
 	github.com/libp2p/go-openssl v0.0.7 // indirect
@@ -39,7 +35,6 @@ require (
 	github.com/multiformats/go-base32 v0.0.3 // indirect
 	github.com/multiformats/go-base36 v0.1.0 // indirect
 	github.com/multiformats/go-multiaddr v0.5.0 // indirect
-	github.com/multiformats/go-multibase v0.0.3 // indirect
 	github.com/multiformats/go-multicodec v0.4.1 // indirect
 	github.com/multiformats/go-varint v0.0.6 // indirect
 	github.com/pkg/errors v0.9.1 // indirect
@@ -52,11 +47,7 @@ require (
 	golang.org/x/crypto v0.0.0-20220411220226-7b82a4e95df4 // indirect
 	golang.org/x/image v0.0.0-20200119044424-58c23975cae1 // indirect
 	golang.org/x/sys v0.0.0-20220422013727-9388b58f7150 // indirect
-	google.golang.org/genproto v0.0.0-20200825200019-8632dd797987 // indirect
+	gopkg.in/check.v1 v1.0.0-20190902080502-41f04d3bba15 // indirect
-	google.golang.org/grpc v1.48.0 // indirect
-	google.golang.org/protobuf v1.27.1 // indirect
 	gopkg.in/yaml.v2 v2.4.0 // indirect
 	lukechampine.com/blake3 v1.1.6 // indirect
 )
-
-replace github.com/textileio/go-threads => github.com/anytypeio/go-threads v1.1.0-rc1.0.20220223104843-a67245cee80e

service/sync/drpcserver/drpcserver.go

@@ -5,8 +5,9 @@ import (
 	"github.com/anytypeio/go-anytype-infrastructure-experiments/app"
 	"github.com/anytypeio/go-anytype-infrastructure-experiments/app/logger"
 	"github.com/anytypeio/go-anytype-infrastructure-experiments/config"
+	"github.com/anytypeio/go-anytype-infrastructure-experiments/service/sync/message"
+	"github.com/anytypeio/go-anytype-infrastructure-experiments/service/sync/syncpb"
 	"github.com/anytypeio/go-anytype-infrastructure-experiments/service/sync/transport"
-	"github.com/anytypeio/go-anytype-infrastructure-experiments/syncproto"
 	"github.com/gogo/protobuf/proto"
 	"go.uber.org/zap"
 	"io"
@@ -14,6 +15,7 @@ import (
 	"storj.io/drpc"
 	"storj.io/drpc/drpcserver"
 	"strings"
+	"sync"
 	"time"
 )
 
@@ -27,6 +29,9 @@ func New() DRPCServer {
 
 type DRPCServer interface {
 	app.ComponentRunnable
+
+	SendMessage(peerId string, msg *syncpb.SyncContent)
+	BroadcastMessage(msg *syncpb.SyncContent)
 }
 
 type drpcServer struct {
@@ -34,12 +39,15 @@ type drpcServer struct {
 	drpcServer     *drpcserver.Server
 	transport      transport.Service
 	listeners      []transport.ContextListener
+	messageService message.Service
+
 	cancel func()
 }
 
 func (s *drpcServer) Init(ctx context.Context, a *app.App) (err error) {
 	s.config = a.MustComponent(config.CName).(*config.Config).GrpcServer
 	s.transport = a.MustComponent(transport.CName).(transport.Service)
+	s.messageService = a.MustComponent(message.CName).(message.Service)
 	return nil
 }
 
@@ -114,23 +122,22 @@ func (s *drpcServer) HandleRPC(stream drpc.Stream, rpc string) (err error) {
 	if err != nil {
 		return
 	}
-	l := log.With(zap.String("peer", sc.RemotePeer().String()))
+	peerId := sc.RemotePeer().String()
+	l := log.With(zap.String("peer", peerId))
 	l.Info("stream opened")
 	defer func() {
 		l.Info("stream closed", zap.Error(err))
 	}()
-	for {
+
-		msg := &syncproto.SyncMessage{}
+	ch := s.messageService.RegisterMessageSender(peerId)
-		if err = stream.MsgRecv(msg, enc{}); err != nil {
+	defer s.messageService.UnregisterMessageSender(peerId)
-			if err == io.EOF {
+
-				return
+	wg := &sync.WaitGroup{}
-			}
+	wg.Add(2)
-		}
+	go s.sendMessages(stream, wg, ch)
-		//log.Debug("receive msg", zap.Int("seq", int(msg.Seq)))
+	go s.receiveMessages(stream, wg, peerId)
-		if err = stream.MsgSend(msg, enc{}); err != nil {
+	wg.Wait()
-			return
+
-		}
-	}
 	return nil
 }
 
@@ -146,6 +153,33 @@ func (s *drpcServer) Close(ctx context.Context) (err error) {
 	return
 }
 
+func (s *drpcServer) sendMessages(stream drpc.Stream, wg *sync.WaitGroup, ch chan *syncpb.SyncContent) {
+	defer wg.Done()
+	for {
+		select {
+		case msg := <-ch:
+			if err := stream.MsgSend(msg, enc{}); err != nil {
+				return
+			}
+		case <-stream.Context().Done():
+			return
+		}
+	}
+}
+
+func (s *drpcServer) receiveMessages(stream drpc.Stream, wg *sync.WaitGroup, peerId string) {
+	defer wg.Done()
+	for {
+		msg := &syncpb.SyncContent{}
+		if err := stream.MsgRecv(msg, enc{}); err != nil {
+			if err == io.EOF {
+				return
+			}
+		}
+		s.messageService.HandleMessage(peerId, msg)
+	}
+}
+
 type enc struct{}
 
 func (e enc) Marshal(msg drpc.Message) ([]byte, error) {

service/sync/message/service.go

@@ -0,0 +1,53 @@
+package message
+
+import (
+	"context"
+	"github.com/anytypeio/go-anytype-infrastructure-experiments/app"
+	"github.com/anytypeio/go-anytype-infrastructure-experiments/service/sync/syncpb"
+)
+
+const CName = "Service"
+
+type service struct {
+}
+
+func NewMessageService() app.Component {
+	return &service{}
+}
+
+type Service interface {
+	RegisterMessageSender(peerId string) chan *syncpb.SyncContent
+	UnregisterMessageSender(peerId string)
+	HandleMessage(peerId string, msg *syncpb.SyncContent)
+}
+
+func (c *service) Init(ctx context.Context, a *app.App) (err error) {
+	return nil
+}
+
+func (c *service) Name() (name string) {
+	return CName
+}
+
+func (c *service) Run(ctx context.Context) (err error) {
+	return nil
+}
+
+func (c *service) Close(ctx context.Context) (err error) {
+	return nil
+}
+
+func (c *service) RegisterMessageSender(peerId string) chan *syncpb.SyncContent {
+	//TODO implement me
+	panic("implement me")
+}
+
+func (c *service) UnregisterMessageSender(peerId string) chan *syncpb.SyncContent {
+	//TODO implement me
+	panic("implement me")
+}
+
+func (c *service) HandleMessage(peerId string, msg *syncpb.SyncContent) {
+	//TODO implement me
+	panic("implement me")
+}