Inject deps for test sync service

commonspace/sync/deps.go

@@ -0,0 +1,10 @@
+package sync
+
+type SyncDeps struct {
+	HeadUpdateHandler HeadUpdateHandler
+	HeadUpdateSender  HeadUpdateSender
+	ResponseHandler   ResponseHandler
+	RequestHandler    RequestHandler
+	RequestSender     RequestSender
+	MergeFilter       MergeFilterFunc
+}

commonspace/sync/requestmanager.go

@@ -1,9 +1,7 @@
 package sync
 
 import (
-	"context"
 	"strings"
-	"sync"
 
 	"github.com/gogo/protobuf/proto"
 	"go.uber.org/zap"
@@ -53,11 +51,15 @@ type requestManager struct {
 	requestHandler  RequestHandler
 	responseHandler ResponseHandler
 	requestSender   RequestSender
-	currentRequests map[string]struct{}
+}
-	mx              sync.Mutex
+
-	ctx             context.Context
+func NewRequestManager(deps SyncDeps) RequestManager {
-	cancel          context.CancelFunc
+	return &requestManager{
-	wait            chan struct{}
+		requestPool:     NewRequestPool(),
+		requestHandler:  deps.RequestHandler,
+		responseHandler: deps.ResponseHandler,
+		requestSender:   deps.RequestSender,
+	}
 }
 
 func (r *requestManager) QueueRequest(rq Request) error {

commonspace/sync/sync.go

@@ -22,28 +22,25 @@ type SyncService interface {
 type MergeFilterFunc func(ctx context.Context, msg drpc.Message, q *mb.MB[drpc.Message]) error
 
 type syncService struct {
-	// sendQueue is a multiqueue: peerId -> queue
-	// this queue exists for sending head updates
 	sendQueueProvider multiqueue.QueueProvider[drpc.Message]
-	// receiveQueue is a multiqueue: objectId -> queue
+	receiveQueue      multiqueue.MultiQueue[drpc.Message]
-	// this queue exists for receiving head updates
+	manager           RequestManager
-	receiveQueue multiqueue.MultiQueue[drpc.Message]
+	handler           HeadUpdateHandler
-	// manager is a Request manager which works with both incoming and outgoing requests
+	sender            HeadUpdateSender
-	manager RequestManager
+	mergeFilter       MergeFilterFunc
-	// handler checks if head update is relevant and then queues Request intent if necessary
+	ctx               context.Context
-	handler HeadUpdateHandler
+	cancel            context.CancelFunc
-	// sender sends head updates to peers
-	sender      HeadUpdateSender
-	mergeFilter MergeFilterFunc
-	ctx         context.Context
-	cancel      context.CancelFunc
 }
 
-func NewSyncService() SyncService {
+func NewSyncService(deps SyncDeps) SyncService {
 	s := &syncService{}
 	s.ctx, s.cancel = context.WithCancel(context.Background())
 	s.sendQueueProvider = multiqueue.NewQueueProvider[drpc.Message](100, s.handleOutgoingMessage)
 	s.receiveQueue = multiqueue.New[drpc.Message](s.handleIncomingMessage, 100)
+	s.sender = deps.HeadUpdateSender
+	s.handler = deps.HeadUpdateHandler
+	s.mergeFilter = deps.MergeFilter
+	s.manager = NewRequestManager(deps)
 	return s
 }