Fix objectsync etc
This commit is contained in:
mcrakhman
Mikhail Iudin
parent
c64921a249
commit
dc910d51c5
@ -5,6 +5,7 @@ type ConfigGetter interface {
|
||||
}
|
||||
|
||||
type Config struct {
|
||||
GCTTL int `yaml:"gcTTL"`
|
||||
SyncPeriod int `yaml:"syncPeriod"`
|
||||
GCTTL int `yaml:"gcTTL"`
|
||||
SyncPeriod int `yaml:"syncPeriod"`
|
||||
TreeInMemoryData bool `yaml:"treeInMemoryData"`
|
||||
}
|
||||
|
||||
@ -1,5 +1,5 @@
|
||||
// Code generated by MockGen. DO NOT EDIT.
|
||||
// Source: github.com/anytypeio/any-sync/commonspace/object/tree/synctree (interfaces: SyncClient,SyncTree,ReceiveQueue,HeadNotifiable)
|
||||
// Source: github.com/anytypeio/any-sync/commonspace/object/tree/synctree (interfaces: SyncTree,ReceiveQueue,HeadNotifiable)
|
||||
|
||||
// Package mock_synctree is a generated GoMock package.
|
||||
package mock_synctree
|
||||
@ -18,115 +18,6 @@ import (
|
||||
gomock "github.com/golang/mock/gomock"
|
||||
)
|
||||
|
||||
// MockSyncClient is a mock of SyncClient interface.
|
||||
type MockSyncClient struct {
|
||||
ctrl *gomock.Controller
|
||||
recorder *MockSyncClientMockRecorder
|
||||
}
|
||||
|
||||
// MockSyncClientMockRecorder is the mock recorder for MockSyncClient.
|
||||
type MockSyncClientMockRecorder struct {
|
||||
mock *MockSyncClient
|
||||
}
|
||||
|
||||
// NewMockSyncClient creates a new mock instance.
|
||||
func NewMockSyncClient(ctrl *gomock.Controller) *MockSyncClient {
|
||||
mock := &MockSyncClient{ctrl: ctrl}
|
||||
mock.recorder = &MockSyncClientMockRecorder{mock}
|
||||
return mock
|
||||
}
|
||||
|
||||
// EXPECT returns an object that allows the caller to indicate expected use.
|
||||
func (m *MockSyncClient) EXPECT() *MockSyncClientMockRecorder {
|
||||
return m.recorder
|
||||
}
|
||||
|
||||
// Broadcast mocks base method.
|
||||
func (m *MockSyncClient) Broadcast(arg0 context.Context, arg1 *treechangeproto.TreeSyncMessage) error {
|
||||
m.ctrl.T.Helper()
|
||||
ret := m.ctrl.Call(m, "Broadcast", arg0, arg1)
|
||||
ret0, _ := ret[0].(error)
|
||||
return ret0
|
||||
}
|
||||
|
||||
// Broadcast indicates an expected call of Broadcast.
|
||||
func (mr *MockSyncClientMockRecorder) Broadcast(arg0, arg1 interface{}) *gomock.Call {
|
||||
mr.mock.ctrl.T.Helper()
|
||||
return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Broadcast", reflect.TypeOf((*MockSyncClient)(nil).Broadcast), arg0, arg1)
|
||||
}
|
||||
|
||||
// CreateFullSyncRequest mocks base method.
|
||||
func (m *MockSyncClient) CreateFullSyncRequest(arg0 objecttree.ObjectTree, arg1, arg2 []string) (*treechangeproto.TreeSyncMessage, error) {
|
||||
m.ctrl.T.Helper()
|
||||
ret := m.ctrl.Call(m, "CreateFullSyncRequest", arg0, arg1, arg2)
|
||||
ret0, _ := ret[0].(*treechangeproto.TreeSyncMessage)
|
||||
ret1, _ := ret[1].(error)
|
||||
return ret0, ret1
|
||||
}
|
||||
|
||||
// CreateFullSyncRequest indicates an expected call of CreateFullSyncRequest.
|
||||
func (mr *MockSyncClientMockRecorder) CreateFullSyncRequest(arg0, arg1, arg2 interface{}) *gomock.Call {
|
||||
mr.mock.ctrl.T.Helper()
|
||||
return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "CreateFullSyncRequest", reflect.TypeOf((*MockSyncClient)(nil).CreateFullSyncRequest), arg0, arg1, arg2)
|
||||
}
|
||||
|
||||
// CreateFullSyncResponse mocks base method.
|
||||
func (m *MockSyncClient) CreateFullSyncResponse(arg0 objecttree.ObjectTree, arg1, arg2 []string) (*treechangeproto.TreeSyncMessage, error) {
|
||||
m.ctrl.T.Helper()
|
||||
ret := m.ctrl.Call(m, "CreateFullSyncResponse", arg0, arg1, arg2)
|
||||
ret0, _ := ret[0].(*treechangeproto.TreeSyncMessage)
|
||||
ret1, _ := ret[1].(error)
|
||||
return ret0, ret1
|
||||
}
|
||||
|
||||
// CreateFullSyncResponse indicates an expected call of CreateFullSyncResponse.
|
||||
func (mr *MockSyncClientMockRecorder) CreateFullSyncResponse(arg0, arg1, arg2 interface{}) *gomock.Call {
|
||||
mr.mock.ctrl.T.Helper()
|
||||
return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "CreateFullSyncResponse", reflect.TypeOf((*MockSyncClient)(nil).CreateFullSyncResponse), arg0, arg1, arg2)
|
||||
}
|
||||
|
||||
// CreateHeadUpdate mocks base method.
|
||||
func (m *MockSyncClient) CreateHeadUpdate(arg0 objecttree.ObjectTree, arg1 []*treechangeproto.RawTreeChangeWithId) *treechangeproto.TreeSyncMessage {
|
||||
m.ctrl.T.Helper()
|
||||
ret := m.ctrl.Call(m, "CreateHeadUpdate", arg0, arg1)
|
||||
ret0, _ := ret[0].(*treechangeproto.TreeSyncMessage)
|
||||
return ret0
|
||||
}
|
||||
|
||||
// CreateHeadUpdate indicates an expected call of CreateHeadUpdate.
|
||||
func (mr *MockSyncClientMockRecorder) CreateHeadUpdate(arg0, arg1 interface{}) *gomock.Call {
|
||||
mr.mock.ctrl.T.Helper()
|
||||
return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "CreateHeadUpdate", reflect.TypeOf((*MockSyncClient)(nil).CreateHeadUpdate), arg0, arg1)
|
||||
}
|
||||
|
||||
// CreateNewTreeRequest mocks base method.
|
||||
func (m *MockSyncClient) CreateNewTreeRequest() *treechangeproto.TreeSyncMessage {
|
||||
m.ctrl.T.Helper()
|
||||
ret := m.ctrl.Call(m, "CreateNewTreeRequest")
|
||||
ret0, _ := ret[0].(*treechangeproto.TreeSyncMessage)
|
||||
return ret0
|
||||
}
|
||||
|
||||
// CreateNewTreeRequest indicates an expected call of CreateNewTreeRequest.
|
||||
func (mr *MockSyncClientMockRecorder) CreateNewTreeRequest() *gomock.Call {
|
||||
mr.mock.ctrl.T.Helper()
|
||||
return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "CreateNewTreeRequest", reflect.TypeOf((*MockSyncClient)(nil).CreateNewTreeRequest))
|
||||
}
|
||||
|
||||
// SendWithReply mocks base method.
|
||||
func (m *MockSyncClient) SendWithReply(arg0 context.Context, arg1 string, arg2 *treechangeproto.TreeSyncMessage, arg3 string) error {
|
||||
m.ctrl.T.Helper()
|
||||
ret := m.ctrl.Call(m, "SendWithReply", arg0, arg1, arg2, arg3)
|
||||
ret0, _ := ret[0].(error)
|
||||
return ret0
|
||||
}
|
||||
|
||||
// SendWithReply indicates an expected call of SendWithReply.
|
||||
func (mr *MockSyncClientMockRecorder) SendWithReply(arg0, arg1, arg2, arg3 interface{}) *gomock.Call {
|
||||
mr.mock.ctrl.T.Helper()
|
||||
return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "SendWithReply", reflect.TypeOf((*MockSyncClient)(nil).SendWithReply), arg0, arg1, arg2, arg3)
|
||||
}
|
||||
|
||||
// MockSyncTree is a mock of SyncTree interface.
|
||||
type MockSyncTree struct {
|
||||
ctrl *gomock.Controller
|
||||
|
||||
@ -1,3 +1,4 @@
|
||||
//go:generate mockgen -destination mock_synctree/mock_synctree.go github.com/anytypeio/any-sync/commonspace/object/tree/synctree SyncTree,ReceiveQueue,HeadNotifiable
|
||||
package synctree
|
||||
|
||||
import (
|
||||
@ -43,7 +44,7 @@ type SyncTree interface {
|
||||
type syncTree struct {
|
||||
objecttree.ObjectTree
|
||||
synchandler.SyncHandler
|
||||
syncClient SyncClient
|
||||
syncClient objectsync.SyncClient
|
||||
syncStatus syncstatus.StatusUpdater
|
||||
notifiable HeadNotifiable
|
||||
listener updatelistener.UpdateListener
|
||||
@ -54,7 +55,7 @@ type syncTree struct {
|
||||
|
||||
var log = logger.NewNamed("common.commonspace.synctree")
|
||||
|
||||
var createSyncClient = NewSyncClient
|
||||
var createSyncClient = objectsync.NewSyncClient
|
||||
|
||||
type ResponsiblePeersGetter interface {
|
||||
GetResponsiblePeers(ctx context.Context) (peers []peer.Peer, err error)
|
||||
@ -101,7 +102,7 @@ func buildSyncTree(ctx context.Context, isFirstBuild bool, deps BuildDeps) (t Sy
|
||||
syncClient := createSyncClient(
|
||||
deps.SpaceId,
|
||||
deps.ObjectSync.MessagePool(),
|
||||
sharedFactory)
|
||||
objectsync.GetRequestFactory())
|
||||
syncTree := &syncTree{
|
||||
ObjectTree: objTree,
|
||||
syncClient: syncClient,
|
||||
|
||||
@ -4,11 +4,11 @@ import (
|
||||
"context"
|
||||
"github.com/anytypeio/any-sync/commonspace/object/tree/objecttree"
|
||||
"github.com/anytypeio/any-sync/commonspace/object/tree/objecttree/mock_objecttree"
|
||||
"github.com/anytypeio/any-sync/commonspace/object/tree/synctree/mock_synctree"
|
||||
"github.com/anytypeio/any-sync/commonspace/object/tree/synctree/updatelistener"
|
||||
"github.com/anytypeio/any-sync/commonspace/object/tree/synctree/updatelistener/mock_updatelistener"
|
||||
"github.com/anytypeio/any-sync/commonspace/object/tree/treechangeproto"
|
||||
"github.com/anytypeio/any-sync/commonspace/objectsync"
|
||||
"github.com/anytypeio/any-sync/commonspace/objectsync/mock_objectsync"
|
||||
"github.com/anytypeio/any-sync/commonspace/syncstatus"
|
||||
"github.com/anytypeio/any-sync/nodeconf"
|
||||
"github.com/golang/mock/gomock"
|
||||
@ -18,7 +18,7 @@ import (
|
||||
|
||||
type syncTreeMatcher struct {
|
||||
objTree objecttree.ObjectTree
|
||||
client SyncClient
|
||||
client objectsync.SyncClient
|
||||
listener updatelistener.UpdateListener
|
||||
}
|
||||
|
||||
@ -34,8 +34,8 @@ func (s syncTreeMatcher) String() string {
|
||||
return ""
|
||||
}
|
||||
|
||||
func syncClientFuncCreator(client SyncClient) func(spaceId string, factory RequestFactory, objectSync objectsync.ObjectSync, configuration nodeconf.Configuration) SyncClient {
|
||||
return func(spaceId string, factory RequestFactory, objectSync objectsync.ObjectSync, configuration nodeconf.Configuration) SyncClient {
|
||||
func syncClientFuncCreator(client objectsync.SyncClient) func(spaceId string, factory objectsync.RequestFactory, objectSync objectsync.ObjectSync, configuration nodeconf.Configuration) objectsync.SyncClient {
|
||||
return func(spaceId string, factory objectsync.RequestFactory, objectSync objectsync.ObjectSync, configuration nodeconf.Configuration) objectsync.SyncClient {
|
||||
return client
|
||||
}
|
||||
}
|
||||
@ -46,7 +46,7 @@ func Test_BuildSyncTree(t *testing.T) {
|
||||
defer ctrl.Finish()
|
||||
|
||||
updateListenerMock := mock_updatelistener.NewMockUpdateListener(ctrl)
|
||||
syncClientMock := mock_synctree.NewMockSyncClient(ctrl)
|
||||
syncClientMock := mock_objectsync.NewMockSyncClient(ctrl)
|
||||
objTreeMock := newTestObjMock(mock_objecttree.NewMockObjectTree(ctrl))
|
||||
tr := &syncTree{
|
||||
ObjectTree: objTreeMock,
|
||||
|
||||
@ -4,6 +4,7 @@ import (
|
||||
"context"
|
||||
"github.com/anytypeio/any-sync/commonspace/object/tree/objecttree"
|
||||
"github.com/anytypeio/any-sync/commonspace/object/tree/treechangeproto"
|
||||
"github.com/anytypeio/any-sync/commonspace/objectsync"
|
||||
"github.com/anytypeio/any-sync/commonspace/objectsync/synchandler"
|
||||
"github.com/anytypeio/any-sync/commonspace/spacesyncproto"
|
||||
"github.com/anytypeio/any-sync/commonspace/syncstatus"
|
||||
@ -15,7 +16,7 @@ import (
|
||||
|
||||
type syncTreeHandler struct {
|
||||
objTree objecttree.ObjectTree
|
||||
syncClient SyncClient
|
||||
syncClient objectsync.SyncClient
|
||||
syncStatus syncstatus.StatusUpdater
|
||||
handlerLock sync.Mutex
|
||||
spaceId string
|
||||
@ -24,7 +25,7 @@ type syncTreeHandler struct {
|
||||
|
||||
const maxQueueSize = 5
|
||||
|
||||
func newSyncTreeHandler(spaceId string, objTree objecttree.ObjectTree, syncClient SyncClient, syncStatus syncstatus.StatusUpdater) synchandler.SyncHandler {
|
||||
func newSyncTreeHandler(spaceId string, objTree objecttree.ObjectTree, syncClient objectsync.SyncClient, syncStatus syncstatus.StatusUpdater) synchandler.SyncHandler {
|
||||
return &syncTreeHandler{
|
||||
objTree: objTree,
|
||||
syncClient: syncClient,
|
||||
@ -166,7 +167,7 @@ func (s *syncTreeHandler) handleFullSyncRequest(
|
||||
defer func() {
|
||||
if err != nil {
|
||||
log.With(zap.Error(err)).DebugCtx(ctx, "full sync request finished with error")
|
||||
s.syncClient.SendWithReply(ctx, senderId, treechangeproto.WrapError(err, header), replyId)
|
||||
s.syncClient.SendWithReply(ctx, senderId, treechangeproto.WrapError(treechangeproto.ErrorCodes_FullSyncRequestError, header), replyId)
|
||||
return
|
||||
} else if fullResponse != nil {
|
||||
cnt := fullResponse.Content.GetFullSyncResponse()
|
||||
|
||||
@ -3,13 +3,14 @@ package synctree
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"github.com/anytypeio/any-sync/commonspace/objectsync"
|
||||
"github.com/anytypeio/any-sync/commonspace/objectsync/mock_objectsync"
|
||||
"sync"
|
||||
"testing"
|
||||
|
||||
"github.com/anytypeio/any-sync/app/logger"
|
||||
"github.com/anytypeio/any-sync/commonspace/object/tree/objecttree"
|
||||
"github.com/anytypeio/any-sync/commonspace/object/tree/objecttree/mock_objecttree"
|
||||
"github.com/anytypeio/any-sync/commonspace/object/tree/synctree/mock_synctree"
|
||||
"github.com/anytypeio/any-sync/commonspace/object/tree/treechangeproto"
|
||||
"github.com/anytypeio/any-sync/commonspace/syncstatus"
|
||||
"github.com/golang/mock/gomock"
|
||||
@ -54,7 +55,7 @@ func (t *testObjTreeMock) TryRLock() bool {
|
||||
|
||||
type syncHandlerFixture struct {
|
||||
ctrl *gomock.Controller
|
||||
syncClientMock *mock_synctree.MockSyncClient
|
||||
syncClientMock *mock_objectsync.MockSyncClient
|
||||
objectTreeMock *testObjTreeMock
|
||||
receiveQueueMock ReceiveQueue
|
||||
|
||||
@ -63,7 +64,7 @@ type syncHandlerFixture struct {
|
||||
|
||||
func newSyncHandlerFixture(t *testing.T) *syncHandlerFixture {
|
||||
ctrl := gomock.NewController(t)
|
||||
syncClientMock := mock_synctree.NewMockSyncClient(ctrl)
|
||||
syncClientMock := mock_objectsync.NewMockSyncClient(ctrl)
|
||||
objectTreeMock := newTestObjMock(mock_objecttree.NewMockObjectTree(ctrl))
|
||||
receiveQueue := newReceiveQueue(5)
|
||||
|
||||
@ -109,7 +110,7 @@ func TestSyncHandler_HandleHeadUpdate(t *testing.T) {
|
||||
SnapshotPath: []string{"h1"},
|
||||
}
|
||||
treeMsg := treechangeproto.WrapHeadUpdate(headUpdate, chWithId)
|
||||
objectMsg, _ := marshallTreeMessage(treeMsg, "spaceId", treeId, "")
|
||||
objectMsg, _ := objectsync.MarshallTreeMessage(treeMsg, "spaceId", treeId, "")
|
||||
|
||||
fx.objectTreeMock.EXPECT().Id().AnyTimes().Return(treeId)
|
||||
fx.objectTreeMock.EXPECT().Heads().Return([]string{"h2"}).Times(2)
|
||||
@ -138,7 +139,7 @@ func TestSyncHandler_HandleHeadUpdate(t *testing.T) {
|
||||
SnapshotPath: []string{"h1"},
|
||||
}
|
||||
treeMsg := treechangeproto.WrapHeadUpdate(headUpdate, chWithId)
|
||||
objectMsg, _ := marshallTreeMessage(treeMsg, "spaceId", treeId, "")
|
||||
objectMsg, _ := objectsync.MarshallTreeMessage(treeMsg, "spaceId", treeId, "")
|
||||
|
||||
fx.objectTreeMock.EXPECT().Id().AnyTimes().Return(treeId)
|
||||
fx.objectTreeMock.EXPECT().Heads().Return([]string{"h2"}).AnyTimes()
|
||||
@ -171,7 +172,7 @@ func TestSyncHandler_HandleHeadUpdate(t *testing.T) {
|
||||
SnapshotPath: []string{"h1"},
|
||||
}
|
||||
treeMsg := treechangeproto.WrapHeadUpdate(headUpdate, chWithId)
|
||||
objectMsg, _ := marshallTreeMessage(treeMsg, "spaceId", treeId, "")
|
||||
objectMsg, _ := objectsync.MarshallTreeMessage(treeMsg, "spaceId", treeId, "")
|
||||
|
||||
fx.objectTreeMock.EXPECT().Id().AnyTimes().Return(treeId)
|
||||
fx.objectTreeMock.EXPECT().Heads().Return([]string{"h1"}).AnyTimes()
|
||||
@ -192,7 +193,7 @@ func TestSyncHandler_HandleHeadUpdate(t *testing.T) {
|
||||
SnapshotPath: []string{"h1"},
|
||||
}
|
||||
treeMsg := treechangeproto.WrapHeadUpdate(headUpdate, chWithId)
|
||||
objectMsg, _ := marshallTreeMessage(treeMsg, "spaceId", treeId, "")
|
||||
objectMsg, _ := objectsync.MarshallTreeMessage(treeMsg, "spaceId", treeId, "")
|
||||
|
||||
fx.objectTreeMock.EXPECT().Id().AnyTimes().Return(treeId)
|
||||
fx.objectTreeMock.EXPECT().Heads().Return([]string{"h2"}).AnyTimes()
|
||||
@ -217,7 +218,7 @@ func TestSyncHandler_HandleHeadUpdate(t *testing.T) {
|
||||
SnapshotPath: []string{"h1"},
|
||||
}
|
||||
treeMsg := treechangeproto.WrapHeadUpdate(headUpdate, chWithId)
|
||||
objectMsg, _ := marshallTreeMessage(treeMsg, "spaceId", treeId, "")
|
||||
objectMsg, _ := objectsync.MarshallTreeMessage(treeMsg, "spaceId", treeId, "")
|
||||
|
||||
fx.objectTreeMock.EXPECT().Id().AnyTimes().Return(treeId)
|
||||
fx.objectTreeMock.EXPECT().Heads().Return([]string{"h1"}).AnyTimes()
|
||||
@ -250,7 +251,7 @@ func TestSyncHandler_HandleFullSyncRequest(t *testing.T) {
|
||||
SnapshotPath: []string{"h1"},
|
||||
}
|
||||
treeMsg := treechangeproto.WrapFullRequest(fullSyncRequest, chWithId)
|
||||
objectMsg, _ := marshallTreeMessage(treeMsg, "spaceId", treeId, "")
|
||||
objectMsg, _ := objectsync.MarshallTreeMessage(treeMsg, "spaceId", treeId, "")
|
||||
|
||||
fx.objectTreeMock.EXPECT().Id().AnyTimes().Return(treeId)
|
||||
fx.objectTreeMock.EXPECT().Header().Return(nil)
|
||||
@ -283,7 +284,7 @@ func TestSyncHandler_HandleFullSyncRequest(t *testing.T) {
|
||||
SnapshotPath: []string{"h1"},
|
||||
}
|
||||
treeMsg := treechangeproto.WrapFullRequest(fullSyncRequest, chWithId)
|
||||
objectMsg, _ := marshallTreeMessage(treeMsg, "spaceId", treeId, "")
|
||||
objectMsg, _ := objectsync.MarshallTreeMessage(treeMsg, "spaceId", treeId, "")
|
||||
|
||||
fx.objectTreeMock.EXPECT().
|
||||
Id().AnyTimes().Return(treeId)
|
||||
@ -312,7 +313,7 @@ func TestSyncHandler_HandleFullSyncRequest(t *testing.T) {
|
||||
SnapshotPath: []string{"h1"},
|
||||
}
|
||||
treeMsg := treechangeproto.WrapFullRequest(fullSyncRequest, chWithId)
|
||||
objectMsg, _ := marshallTreeMessage(treeMsg, "spaceId", treeId, "")
|
||||
objectMsg, _ := objectsync.MarshallTreeMessage(treeMsg, "spaceId", treeId, "")
|
||||
objectMsg.RequestId = replyId
|
||||
|
||||
fx.objectTreeMock.EXPECT().
|
||||
@ -339,7 +340,7 @@ func TestSyncHandler_HandleFullSyncRequest(t *testing.T) {
|
||||
SnapshotPath: []string{"h1"},
|
||||
}
|
||||
treeMsg := treechangeproto.WrapFullRequest(fullSyncRequest, chWithId)
|
||||
objectMsg, _ := marshallTreeMessage(treeMsg, "spaceId", treeId, "")
|
||||
objectMsg, _ := objectsync.MarshallTreeMessage(treeMsg, "spaceId", treeId, "")
|
||||
|
||||
fx.objectTreeMock.EXPECT().
|
||||
Id().AnyTimes().Return(treeId)
|
||||
@ -380,7 +381,7 @@ func TestSyncHandler_HandleFullSyncResponse(t *testing.T) {
|
||||
SnapshotPath: []string{"h1"},
|
||||
}
|
||||
treeMsg := treechangeproto.WrapFullResponse(fullSyncResponse, chWithId)
|
||||
objectMsg, _ := marshallTreeMessage(treeMsg, "spaceId", treeId, replyId)
|
||||
objectMsg, _ := objectsync.MarshallTreeMessage(treeMsg, "spaceId", treeId, replyId)
|
||||
|
||||
fx.objectTreeMock.EXPECT().Id().AnyTimes().Return(treeId)
|
||||
fx.objectTreeMock.EXPECT().
|
||||
@ -413,7 +414,7 @@ func TestSyncHandler_HandleFullSyncResponse(t *testing.T) {
|
||||
SnapshotPath: []string{"h1"},
|
||||
}
|
||||
treeMsg := treechangeproto.WrapFullResponse(fullSyncResponse, chWithId)
|
||||
objectMsg, _ := marshallTreeMessage(treeMsg, "spaceId", treeId, replyId)
|
||||
objectMsg, _ := objectsync.MarshallTreeMessage(treeMsg, "spaceId", treeId, replyId)
|
||||
|
||||
fx.objectTreeMock.EXPECT().Id().AnyTimes().Return(treeId)
|
||||
fx.objectTreeMock.EXPECT().
|
||||
|
||||
@ -6,6 +6,7 @@ import (
|
||||
"github.com/anytypeio/any-sync/commonspace/object/tree/objecttree"
|
||||
"github.com/anytypeio/any-sync/commonspace/object/tree/treechangeproto"
|
||||
"github.com/anytypeio/any-sync/commonspace/object/tree/treestorage"
|
||||
"github.com/anytypeio/any-sync/commonspace/objectsync"
|
||||
"github.com/anytypeio/any-sync/commonspace/spacestorage"
|
||||
"github.com/anytypeio/any-sync/net/peer"
|
||||
"github.com/anytypeio/any-sync/net/rpc/rpcerr"
|
||||
@ -46,8 +47,8 @@ func (t treeRemoteGetter) getPeers(ctx context.Context) (peerIds []string, err e
|
||||
}
|
||||
|
||||
func (t treeRemoteGetter) treeRequest(ctx context.Context, peerId string) (msg *treechangeproto.TreeSyncMessage, err error) {
|
||||
newTreeRequest := GetRequestFactory().CreateNewTreeRequest()
|
||||
objMsg, err := marshallTreeMessage(newTreeRequest, t.deps.SpaceId, t.treeId, "")
|
||||
newTreeRequest := objectsync.GetRequestFactory().CreateNewTreeRequest()
|
||||
objMsg, err := objectsync.MarshallTreeMessage(newTreeRequest, t.deps.SpaceId, t.treeId, "")
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
|
||||
@ -7,6 +7,7 @@ import (
|
||||
"github.com/anytypeio/any-sync/commonspace/object/tree/objecttree"
|
||||
"github.com/anytypeio/any-sync/commonspace/object/tree/treechangeproto"
|
||||
"github.com/anytypeio/any-sync/commonspace/object/tree/treestorage"
|
||||
"github.com/anytypeio/any-sync/commonspace/objectsync"
|
||||
"github.com/anytypeio/any-sync/commonspace/objectsync/synchandler"
|
||||
"github.com/anytypeio/any-sync/commonspace/spacesyncproto"
|
||||
"github.com/anytypeio/any-sync/commonspace/syncstatus"
|
||||
@ -84,7 +85,7 @@ type processSyncHandler struct {
|
||||
peerId string
|
||||
aclList list.AclList
|
||||
log *messageLog
|
||||
syncClient SyncClient
|
||||
syncClient objectsync.SyncClient
|
||||
}
|
||||
|
||||
func (p *processSyncHandler) HandleMessage(ctx context.Context, senderId string, request *spacesyncproto.ObjectSyncMessage) (err error) {
|
||||
@ -97,9 +98,9 @@ func (p *processSyncHandler) HandleMessage(ctx context.Context, senderId string,
|
||||
return
|
||||
}
|
||||
if unmarshalled.Content.GetFullSyncResponse() == nil {
|
||||
newTreeRequest := GetRequestFactory().CreateNewTreeRequest()
|
||||
newTreeRequest := objectsync.GetRequestFactory().CreateNewTreeRequest()
|
||||
var objMsg *spacesyncproto.ObjectSyncMessage
|
||||
objMsg, err = marshallTreeMessage(newTreeRequest, request.SpaceId, request.ObjectId, "")
|
||||
objMsg, err = objectsync.MarshallTreeMessage(newTreeRequest, request.SpaceId, request.ObjectId, "")
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
@ -124,8 +125,8 @@ func (p *processSyncHandler) HandleMessage(ctx context.Context, senderId string,
|
||||
}
|
||||
p.SyncHandler = newSyncTreeHandler(request.SpaceId, netTree, p.syncClient, syncstatus.NewNoOpSyncStatus())
|
||||
var objMsg *spacesyncproto.ObjectSyncMessage
|
||||
newTreeRequest := GetRequestFactory().CreateHeadUpdate(netTree, res.Added)
|
||||
objMsg, err = marshallTreeMessage(newTreeRequest, request.SpaceId, request.ObjectId, "")
|
||||
newTreeRequest := objectsync.GetRequestFactory().CreateHeadUpdate(netTree, res.Added)
|
||||
objMsg, err = objectsync.MarshallTreeMessage(newTreeRequest, request.SpaceId, request.ObjectId, "")
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
@ -143,9 +144,9 @@ func newProcessSyncHandler(peerId string, syncHandler synchandler.SyncHandler) *
|
||||
|
||||
func (p *processSyncHandler) manager() *processPeerManager {
|
||||
if p.SyncHandler != nil {
|
||||
return p.SyncHandler.(*syncTreeHandler).syncClient.(*syncClient).PeerManager.(*processPeerManager)
|
||||
return p.SyncHandler.(*syncTreeHandler).syncClient.PeerManager().(*processPeerManager)
|
||||
}
|
||||
return p.syncClient.(*syncClient).PeerManager.(*processPeerManager)
|
||||
return p.syncClient.PeerManager().(*processPeerManager)
|
||||
}
|
||||
|
||||
func (p *processSyncHandler) tree() *broadcastTree {
|
||||
@ -229,7 +230,7 @@ func (m *processPeerManager) GetResponsiblePeers(ctx context.Context) (peers []p
|
||||
|
||||
type broadcastTree struct {
|
||||
objecttree.ObjectTree
|
||||
SyncClient
|
||||
objectsync.SyncClient
|
||||
}
|
||||
|
||||
func (b *broadcastTree) AddRawChanges(ctx context.Context, changes objecttree.RawChangesPayload) (objecttree.AddResult, error) {
|
||||
@ -243,8 +244,8 @@ func (b *broadcastTree) AddRawChanges(ctx context.Context, changes objecttree.Ra
|
||||
}
|
||||
|
||||
func createSyncHandler(peerId, spaceId string, objTree objecttree.ObjectTree, log *messageLog) *processSyncHandler {
|
||||
factory := GetRequestFactory()
|
||||
syncClient := NewSyncClient(spaceId, newProcessPeerManager(peerId, log), factory)
|
||||
factory := objectsync.GetRequestFactory()
|
||||
syncClient := objectsync.NewSyncClient(spaceId, newProcessPeerManager(peerId, log), factory)
|
||||
netTree := &broadcastTree{
|
||||
ObjectTree: objTree,
|
||||
SyncClient: syncClient,
|
||||
@ -254,8 +255,8 @@ func createSyncHandler(peerId, spaceId string, objTree objecttree.ObjectTree, lo
|
||||
}
|
||||
|
||||
func createEmptySyncHandler(peerId, spaceId string, aclList list.AclList, log *messageLog) *processSyncHandler {
|
||||
factory := GetRequestFactory()
|
||||
syncClient := NewSyncClient(spaceId, newProcessPeerManager(peerId, log), factory)
|
||||
factory := objectsync.GetRequestFactory()
|
||||
syncClient := objectsync.NewSyncClient(spaceId, newProcessPeerManager(peerId, log), factory)
|
||||
|
||||
batcher := mb.New[processMsg](0)
|
||||
return &processSyncHandler{
|
||||
|
||||
14
commonspace/object/tree/treechangeproto/errors.go
Normal file
14
commonspace/object/tree/treechangeproto/errors.go
Normal file
@ -0,0 +1,14 @@
|
||||
package treechangeproto
|
||||
|
||||
import (
|
||||
"errors"
|
||||
"github.com/anytypeio/any-sync/net/rpc/rpcerr"
|
||||
)
|
||||
|
||||
var (
|
||||
errGroup = rpcerr.ErrGroup(ErrorCodes_ErrorOffset)
|
||||
|
||||
ErrUnexpected = errGroup.Register(errors.New("unexpected error"), uint64(ErrorCodes_Unexpected))
|
||||
ErrGetTree = errGroup.Register(errors.New("tree not found"), uint64(ErrorCodes_GetTreeError))
|
||||
ErrFullSync = errGroup.Register(errors.New("full sync request error"), uint64(ErrorCodes_FullSyncRequestError))
|
||||
)
|
||||
@ -56,6 +56,13 @@ message RawTreeChangeWithId {
|
||||
string id = 2;
|
||||
}
|
||||
|
||||
enum ErrorCodes {
|
||||
Unexpected = 0;
|
||||
GetTreeError = 1;
|
||||
FullSyncRequestError = 2;
|
||||
ErrorOffset = 400;
|
||||
}
|
||||
|
||||
message TreeSyncMessage {
|
||||
TreeSyncContentValue content = 1;
|
||||
RawTreeChangeWithId rootChange = 2;
|
||||
|
||||
@ -27,10 +27,10 @@ func WrapFullResponse(response *TreeFullSyncResponse, rootChange *RawTreeChangeW
|
||||
}
|
||||
}
|
||||
|
||||
func WrapError(err error, rootChange *RawTreeChangeWithId) *TreeSyncMessage {
|
||||
func WrapError(code ErrorCodes, rootChange *RawTreeChangeWithId) *TreeSyncMessage {
|
||||
return &TreeSyncMessage{
|
||||
Content: &TreeSyncContentValue{
|
||||
Value: &TreeSyncContentValue_ErrorResponse{ErrorResponse: &TreeErrorResponse{Error: err.Error()}},
|
||||
Value: &TreeSyncContentValue_ErrorResponse{ErrorResponse: &TreeErrorResponse{ErrCode: uint64(code)}},
|
||||
},
|
||||
RootChange: rootChange,
|
||||
}
|
||||
|
||||
@ -22,6 +22,37 @@ var _ = math.Inf
|
||||
// proto package needs to be updated.
|
||||
const _ = proto.GoGoProtoPackageIsVersion3 // please upgrade the proto package
|
||||
|
||||
type ErrorCodes int32
|
||||
|
||||
const (
|
||||
ErrorCodes_Unexpected ErrorCodes = 0
|
||||
ErrorCodes_GetTreeError ErrorCodes = 1
|
||||
ErrorCodes_FullSyncRequestError ErrorCodes = 2
|
||||
ErrorCodes_ErrorOffset ErrorCodes = 400
|
||||
)
|
||||
|
||||
var ErrorCodes_name = map[int32]string{
|
||||
0: "Unexpected",
|
||||
1: "GetTreeError",
|
||||
2: "FullSyncRequestError",
|
||||
400: "ErrorOffset",
|
||||
}
|
||||
|
||||
var ErrorCodes_value = map[string]int32{
|
||||
"Unexpected": 0,
|
||||
"GetTreeError": 1,
|
||||
"FullSyncRequestError": 2,
|
||||
"ErrorOffset": 400,
|
||||
}
|
||||
|
||||
func (x ErrorCodes) String() string {
|
||||
return proto.EnumName(ErrorCodes_name, int32(x))
|
||||
}
|
||||
|
||||
func (ErrorCodes) EnumDescriptor() ([]byte, []int) {
|
||||
return fileDescriptor_5033f0301ef9b772, []int{0}
|
||||
}
|
||||
|
||||
// RootChange is a root of a tree
|
||||
type RootChange struct {
|
||||
// AclHeadId is a cid of latest acl record at the time of tree creation
|
||||
@ -788,6 +819,7 @@ func (m *TreeChangeInfo) GetChangePayload() []byte {
|
||||
}
|
||||
|
||||
func init() {
|
||||
proto.RegisterEnum("treechange.ErrorCodes", ErrorCodes_name, ErrorCodes_value)
|
||||
proto.RegisterType((*RootChange)(nil), "treechange.RootChange")
|
||||
proto.RegisterType((*TreeChange)(nil), "treechange.TreeChange")
|
||||
proto.RegisterType((*RawTreeChange)(nil), "treechange.RawTreeChange")
|
||||
@ -806,50 +838,53 @@ func init() {
|
||||
}
|
||||
|
||||
var fileDescriptor_5033f0301ef9b772 = []byte{
|
||||
// 681 bytes of a gzipped FileDescriptorProto
|
||||
0x1f, 0x8b, 0x08, 0x00, 0x00, 0x00, 0x00, 0x00, 0x02, 0xff, 0xc4, 0x55, 0x4d, 0x6f, 0xd3, 0x40,
|
||||
0x10, 0xf5, 0x3a, 0x69, 0xdd, 0x4c, 0xd3, 0x16, 0xb6, 0x3d, 0x58, 0x15, 0x18, 0xcb, 0x42, 0x90,
|
||||
0x0b, 0xad, 0x54, 0x4e, 0x20, 0xa4, 0x8a, 0x86, 0x96, 0x44, 0x15, 0xa8, 0xda, 0x96, 0x22, 0x71,
|
||||
0xdb, 0xda, 0xdb, 0xc6, 0x28, 0xf1, 0x1a, 0xef, 0x86, 0x2a, 0x3f, 0x80, 0x0b, 0x48, 0x15, 0x7f,
|
||||
0x89, 0x1b, 0xc7, 0x1e, 0x7b, 0x44, 0xcd, 0x1f, 0x41, 0xde, 0x8d, 0xe3, 0x8f, 0xe4, 0xd0, 0x5b,
|
||||
0x2f, 0x4e, 0xe6, 0x79, 0xe6, 0xcd, 0x9b, 0xb7, 0x1f, 0x86, 0x5d, 0x9f, 0x0f, 0x06, 0x3c, 0x12,
|
||||
0x31, 0xf5, 0xd9, 0x36, 0x3f, 0xfb, 0xca, 0x7c, 0xb9, 0x2d, 0x13, 0xc6, 0xd4, 0xc3, 0xef, 0xd1,
|
||||
0xe8, 0x82, 0xc5, 0x09, 0x97, 0x7c, 0x5b, 0x3d, 0x45, 0x01, 0xde, 0x52, 0x08, 0x86, 0x1c, 0xf1,
|
||||
0x6e, 0x10, 0x00, 0xe1, 0x5c, 0xb6, 0x55, 0x88, 0x1f, 0x41, 0x83, 0xfa, 0xfd, 0x0e, 0xa3, 0x41,
|
||||
0x37, 0xb0, 0x91, 0x8b, 0x5a, 0x0d, 0x92, 0x03, 0xd8, 0x06, 0x4b, 0x75, 0xed, 0x06, 0xb6, 0xa9,
|
||||
0xde, 0x65, 0x21, 0x76, 0x00, 0x34, 0xe1, 0xc9, 0x28, 0x66, 0x76, 0x4d, 0xbd, 0x2c, 0x20, 0x29,
|
||||
0xaf, 0x0c, 0x07, 0x4c, 0x48, 0x3a, 0x88, 0xed, 0xba, 0x8b, 0x5a, 0x35, 0x92, 0x03, 0x18, 0x43,
|
||||
0x5d, 0x30, 0x16, 0xd8, 0x0b, 0x2e, 0x6a, 0x35, 0x89, 0xfa, 0x8f, 0x37, 0x61, 0x29, 0x0c, 0x58,
|
||||
0x24, 0x43, 0x39, 0xb2, 0x17, 0x15, 0x3e, 0x8d, 0xf1, 0x53, 0x58, 0xd1, 0xdc, 0x47, 0x74, 0xd4,
|
||||
0xe7, 0x34, 0xb0, 0x2d, 0x95, 0x50, 0x06, 0xbd, 0x2b, 0x13, 0xe0, 0x24, 0x61, 0x6c, 0x32, 0x9a,
|
||||
0x0b, 0xcb, 0xe9, 0xdc, 0x7a, 0x14, 0x61, 0x23, 0xb7, 0xd6, 0x6a, 0x90, 0x22, 0x54, 0x1e, 0xde,
|
||||
0xac, 0x0e, 0xff, 0x0c, 0x56, 0x45, 0x44, 0x63, 0xd1, 0xe3, 0x72, 0x8f, 0x8a, 0xd4, 0x03, 0x3d,
|
||||
0x66, 0x05, 0x4d, 0xfb, 0x68, 0x1d, 0xe2, 0x1d, 0x95, 0x54, 0x0d, 0xdb, 0x24, 0x45, 0x28, 0xed,
|
||||
0x93, 0x30, 0x1a, 0x1c, 0xb2, 0x51, 0x57, 0xcf, 0xdc, 0x20, 0x39, 0x50, 0xb6, 0x6a, 0xb1, 0x6a,
|
||||
0x55, 0xd1, 0x16, 0xab, 0x62, 0x8b, 0x03, 0x10, 0x8a, 0xe3, 0x89, 0x1a, 0x7b, 0xc9, 0x45, 0xad,
|
||||
0x25, 0x52, 0x40, 0xbc, 0xf7, 0xb0, 0x42, 0xe8, 0x65, 0xc1, 0x12, 0x1b, 0xac, 0x78, 0xe2, 0x20,
|
||||
0x52, 0x5c, 0x59, 0x98, 0x8a, 0x10, 0xe1, 0x45, 0x44, 0xe5, 0x30, 0x61, 0xca, 0x8a, 0x26, 0xc9,
|
||||
0x01, 0xaf, 0x0d, 0xeb, 0x25, 0xa2, 0xcf, 0xa1, 0xec, 0x69, 0xe5, 0x09, 0xbd, 0xd4, 0xd0, 0x84,
|
||||
0x30, 0x07, 0xf0, 0x2a, 0x98, 0x61, 0x66, 0xab, 0x19, 0x06, 0xde, 0x15, 0x82, 0xb5, 0x94, 0xe2,
|
||||
0x78, 0x14, 0xf9, 0x1f, 0x98, 0x10, 0xf4, 0x82, 0xe1, 0xd7, 0x60, 0xf9, 0x3c, 0x92, 0x2c, 0x92,
|
||||
0xaa, 0x7e, 0x79, 0xc7, 0xdd, 0x2a, 0xec, 0xde, 0x2c, 0xbb, 0xad, 0x53, 0x4e, 0x69, 0x7f, 0xc8,
|
||||
0x48, 0x56, 0x80, 0x77, 0x01, 0x92, 0xe9, 0x46, 0x56, 0x7d, 0x96, 0x77, 0x9e, 0x14, 0xcb, 0xe7,
|
||||
0x48, 0x26, 0x85, 0x12, 0xef, 0x8f, 0x09, 0x1b, 0xf3, 0x5a, 0xe0, 0x37, 0x00, 0x3d, 0x46, 0x83,
|
||||
0x4f, 0x71, 0x40, 0x25, 0x9b, 0x08, 0xdb, 0xac, 0x0a, 0xeb, 0x4c, 0x33, 0x3a, 0x06, 0x29, 0xe4,
|
||||
0xe3, 0x43, 0x58, 0x3b, 0x1f, 0xf6, 0xfb, 0x29, 0x2b, 0x61, 0xdf, 0x86, 0x4c, 0xc8, 0x79, 0xe2,
|
||||
0x52, 0x8a, 0x83, 0x72, 0x5a, 0xc7, 0x20, 0xd5, 0x4a, 0xfc, 0x11, 0x1e, 0xe4, 0x90, 0x88, 0x79,
|
||||
0x24, 0xf4, 0x69, 0x9b, 0xe3, 0xd4, 0x41, 0x25, 0xaf, 0x63, 0x90, 0x99, 0x5a, 0xbc, 0x0f, 0x2b,
|
||||
0x2c, 0x49, 0x78, 0x32, 0x25, 0xab, 0x2b, 0xb2, 0xc7, 0x55, 0xb2, 0xfd, 0x62, 0x52, 0xc7, 0x20,
|
||||
0xe5, 0xaa, 0x3d, 0x0b, 0x16, 0xbe, 0xa7, 0x56, 0x79, 0x3f, 0x10, 0xac, 0x96, 0xdd, 0xc0, 0x1b,
|
||||
0xb0, 0x90, 0xba, 0x91, 0x9d, 0x38, 0x1d, 0xe0, 0x57, 0x60, 0x4d, 0x8e, 0x84, 0x6d, 0xba, 0xb5,
|
||||
0xbb, 0x2c, 0x55, 0x96, 0x8f, 0x3d, 0x68, 0x66, 0x47, 0xee, 0x88, 0xca, 0x9e, 0x5d, 0x53, 0xbc,
|
||||
0x25, 0xcc, 0xfb, 0x89, 0x60, 0x7d, 0x8e, 0xa5, 0xf7, 0x23, 0xe6, 0x17, 0xd2, 0x1b, 0xab, 0xba,
|
||||
0x22, 0xf7, 0xa3, 0xe6, 0x05, 0x3c, 0x9c, 0x59, 0xd1, 0xf4, 0x26, 0x60, 0x49, 0xd2, 0xe6, 0x81,
|
||||
0xde, 0xdf, 0x75, 0x92, 0x85, 0xde, 0xa9, 0x5e, 0x50, 0xdd, 0xaf, 0x1b, 0x9d, 0xf3, 0xca, 0x5d,
|
||||
0x8f, 0x66, 0xee, 0xfa, 0x99, 0xdb, 0xd9, 0x9c, 0x73, 0x3b, 0xef, 0xbd, 0xfd, 0x7b, 0xeb, 0xa0,
|
||||
0xeb, 0x5b, 0x07, 0xfd, 0xbb, 0x75, 0xd0, 0xef, 0xb1, 0x63, 0x5c, 0x8f, 0x1d, 0xe3, 0x66, 0xec,
|
||||
0x18, 0x5f, 0x9e, 0xdf, 0xf1, 0xfb, 0x76, 0xb6, 0xa8, 0x7e, 0x5e, 0xfe, 0x0f, 0x00, 0x00, 0xff,
|
||||
0xff, 0x28, 0x34, 0x11, 0x90, 0x11, 0x07, 0x00, 0x00,
|
||||
// 732 bytes of a gzipped FileDescriptorProto
|
||||
0x1f, 0x8b, 0x08, 0x00, 0x00, 0x00, 0x00, 0x00, 0x02, 0xff, 0xc4, 0x55, 0x4f, 0x4f, 0xfb, 0x46,
|
||||
0x10, 0xf5, 0x3a, 0x81, 0x90, 0x49, 0x08, 0xe9, 0xc2, 0xc1, 0x42, 0xad, 0x6b, 0x59, 0x55, 0x1b,
|
||||
0x55, 0x2a, 0x48, 0xf4, 0xd4, 0xaa, 0x12, 0x2a, 0x29, 0x90, 0x08, 0xb5, 0x45, 0xcb, 0x9f, 0x4a,
|
||||
0xdc, 0x16, 0x7b, 0x42, 0x5c, 0x25, 0xb6, 0xeb, 0xdd, 0x94, 0xe6, 0x03, 0xf4, 0xd2, 0x4a, 0x88,
|
||||
0xaf, 0xd4, 0x5b, 0x8f, 0x1c, 0x39, 0xfe, 0x04, 0x5f, 0xe4, 0x27, 0xef, 0xc6, 0x89, 0xed, 0xe4,
|
||||
0xc0, 0x8d, 0x8b, 0xe3, 0x79, 0x9e, 0x79, 0xf3, 0xe6, 0xed, 0x9f, 0xc0, 0xa1, 0x17, 0x8d, 0xc7,
|
||||
0x51, 0x28, 0x62, 0xee, 0xe1, 0x7e, 0x74, 0xfb, 0x3b, 0x7a, 0x72, 0x5f, 0x26, 0x88, 0xea, 0xe1,
|
||||
0x0d, 0x79, 0x78, 0x87, 0x71, 0x12, 0xc9, 0x68, 0x5f, 0x3d, 0x45, 0x0e, 0xde, 0x53, 0x08, 0x85,
|
||||
0x05, 0xe2, 0x3e, 0x13, 0x00, 0x16, 0x45, 0xb2, 0xab, 0x42, 0xfa, 0x29, 0xd4, 0xb9, 0x37, 0xea,
|
||||
0x21, 0xf7, 0xfb, 0xbe, 0x45, 0x1c, 0xd2, 0xa9, 0xb3, 0x05, 0x40, 0x2d, 0xa8, 0xa9, 0xae, 0x7d,
|
||||
0xdf, 0x32, 0xd5, 0xb7, 0x2c, 0xa4, 0x36, 0x80, 0x26, 0xbc, 0x9c, 0xc6, 0x68, 0x55, 0xd4, 0xc7,
|
||||
0x1c, 0x92, 0xf2, 0xca, 0x60, 0x8c, 0x42, 0xf2, 0x71, 0x6c, 0x55, 0x1d, 0xd2, 0xa9, 0xb0, 0x05,
|
||||
0x40, 0x29, 0x54, 0x05, 0xa2, 0x6f, 0xad, 0x39, 0xa4, 0xd3, 0x64, 0xea, 0x9d, 0xee, 0xc2, 0x46,
|
||||
0xe0, 0x63, 0x28, 0x03, 0x39, 0xb5, 0xd6, 0x15, 0x3e, 0x8f, 0xe9, 0x17, 0xb0, 0xa9, 0xb9, 0xcf,
|
||||
0xf9, 0x74, 0x14, 0x71, 0xdf, 0xaa, 0xa9, 0x84, 0x22, 0xe8, 0x3e, 0x98, 0x00, 0x97, 0x09, 0xe2,
|
||||
0x6c, 0x34, 0x07, 0x1a, 0xe9, 0xdc, 0x7a, 0x14, 0x61, 0x11, 0xa7, 0xd2, 0xa9, 0xb3, 0x3c, 0x54,
|
||||
0x1c, 0xde, 0x2c, 0x0f, 0xff, 0x25, 0xb4, 0x44, 0xc8, 0x63, 0x31, 0x8c, 0xe4, 0x11, 0x17, 0xa9,
|
||||
0x07, 0x7a, 0xcc, 0x12, 0x9a, 0xf6, 0xd1, 0x3a, 0xc4, 0x4f, 0x5c, 0x72, 0x35, 0x6c, 0x93, 0xe5,
|
||||
0xa1, 0xb4, 0x4f, 0x82, 0xdc, 0x3f, 0xc3, 0x69, 0x5f, 0xcf, 0x5c, 0x67, 0x0b, 0xa0, 0x68, 0xd5,
|
||||
0x7a, 0xd9, 0xaa, 0xbc, 0x2d, 0xb5, 0x92, 0x2d, 0x36, 0x40, 0x20, 0x2e, 0x66, 0x6a, 0xac, 0x0d,
|
||||
0x87, 0x74, 0x36, 0x58, 0x0e, 0x71, 0x4f, 0x61, 0x93, 0xf1, 0xfb, 0x9c, 0x25, 0x16, 0xd4, 0xe2,
|
||||
0x99, 0x83, 0x44, 0x71, 0x65, 0x61, 0x2a, 0x42, 0x04, 0x77, 0x21, 0x97, 0x93, 0x04, 0x95, 0x15,
|
||||
0x4d, 0xb6, 0x00, 0xdc, 0x2e, 0x6c, 0x17, 0x88, 0x7e, 0x0b, 0xe4, 0x50, 0x2b, 0x4f, 0xf8, 0xbd,
|
||||
0x86, 0x66, 0x84, 0x0b, 0x80, 0xb6, 0xc0, 0x0c, 0x32, 0x5b, 0xcd, 0xc0, 0x77, 0x1f, 0x08, 0x6c,
|
||||
0xa5, 0x14, 0x17, 0xd3, 0xd0, 0xfb, 0x19, 0x85, 0xe0, 0x77, 0x48, 0xbf, 0x87, 0x9a, 0x17, 0x85,
|
||||
0x12, 0x43, 0xa9, 0xea, 0x1b, 0x07, 0xce, 0x5e, 0x6e, 0xf7, 0x66, 0xd9, 0x5d, 0x9d, 0x72, 0xcd,
|
||||
0x47, 0x13, 0x64, 0x59, 0x01, 0x3d, 0x04, 0x48, 0xe6, 0x1b, 0x59, 0xf5, 0x69, 0x1c, 0x7c, 0x9e,
|
||||
0x2f, 0x5f, 0x21, 0x99, 0xe5, 0x4a, 0xdc, 0xff, 0x4c, 0xd8, 0x59, 0xd5, 0x82, 0xfe, 0x00, 0x30,
|
||||
0x44, 0xee, 0x5f, 0xc5, 0x3e, 0x97, 0x38, 0x13, 0xb6, 0x5b, 0x16, 0xd6, 0x9b, 0x67, 0xf4, 0x0c,
|
||||
0x96, 0xcb, 0xa7, 0x67, 0xb0, 0x35, 0x98, 0x8c, 0x46, 0x29, 0x2b, 0xc3, 0x3f, 0x26, 0x28, 0xe4,
|
||||
0x2a, 0x71, 0x29, 0xc5, 0x49, 0x31, 0xad, 0x67, 0xb0, 0x72, 0x25, 0xfd, 0x05, 0xda, 0x0b, 0x48,
|
||||
0xc4, 0x51, 0x28, 0xf4, 0x69, 0x5b, 0xe1, 0xd4, 0x49, 0x29, 0xaf, 0x67, 0xb0, 0xa5, 0x5a, 0x7a,
|
||||
0x0c, 0x9b, 0x98, 0x24, 0x51, 0x32, 0x27, 0xab, 0x2a, 0xb2, 0xcf, 0xca, 0x64, 0xc7, 0xf9, 0xa4,
|
||||
0x9e, 0xc1, 0x8a, 0x55, 0x47, 0x35, 0x58, 0xfb, 0x33, 0xb5, 0xca, 0xfd, 0x9b, 0x40, 0xab, 0xe8,
|
||||
0x06, 0xdd, 0x81, 0xb5, 0xd4, 0x8d, 0xec, 0xc4, 0xe9, 0x80, 0x7e, 0x07, 0xb5, 0xd9, 0x91, 0xb0,
|
||||
0x4c, 0xa7, 0xf2, 0x96, 0xa5, 0xca, 0xf2, 0xa9, 0x0b, 0xcd, 0xec, 0xc8, 0x9d, 0x73, 0x39, 0xb4,
|
||||
0x2a, 0x8a, 0xb7, 0x80, 0xb9, 0xff, 0x10, 0xd8, 0x5e, 0x61, 0xe9, 0xfb, 0x88, 0xf9, 0x97, 0xe8,
|
||||
0x8d, 0x55, 0x5e, 0x91, 0xf7, 0x51, 0xf3, 0x0d, 0x7c, 0xb2, 0xb4, 0xa2, 0xe9, 0x4d, 0x80, 0x49,
|
||||
0xd2, 0x8d, 0x7c, 0xbd, 0xbf, 0xab, 0x2c, 0x0b, 0xdd, 0x6b, 0xbd, 0xa0, 0xba, 0x5f, 0x3f, 0x1c,
|
||||
0x44, 0xa5, 0xbb, 0x9e, 0x2c, 0xdd, 0xf5, 0x4b, 0xb7, 0xb3, 0xb9, 0xe2, 0x76, 0xfe, 0xfa, 0x06,
|
||||
0x40, 0x49, 0x48, 0x9b, 0x08, 0xda, 0x02, 0xb8, 0x0a, 0xf1, 0xaf, 0x18, 0x3d, 0x89, 0x7e, 0xdb,
|
||||
0xa0, 0x6d, 0x68, 0x9e, 0xa2, 0x9c, 0xeb, 0x6c, 0x13, 0x6a, 0xc1, 0x4e, 0x69, 0x31, 0xf5, 0x17,
|
||||
0x93, 0xb6, 0xa1, 0xa1, 0x5e, 0x7f, 0x1d, 0x0c, 0x04, 0xca, 0xf6, 0x63, 0xe5, 0xe8, 0xc7, 0xff,
|
||||
0x5f, 0x6c, 0xf2, 0xf4, 0x62, 0x93, 0x0f, 0x2f, 0x36, 0x79, 0x7c, 0xb5, 0x8d, 0xa7, 0x57, 0xdb,
|
||||
0x78, 0x7e, 0xb5, 0x8d, 0x9b, 0xaf, 0xde, 0xf8, 0xdf, 0x79, 0xbb, 0xae, 0x7e, 0xbe, 0xfd, 0x18,
|
||||
0x00, 0x00, 0xff, 0xff, 0xb5, 0x68, 0xf7, 0x3f, 0x6d, 0x07, 0x00, 0x00,
|
||||
}
|
||||
|
||||
func (m *RootChange) Marshal() (dAtA []byte, err error) {
|
||||
|
||||
@ -9,9 +9,7 @@ import (
|
||||
reflect "reflect"
|
||||
|
||||
app "github.com/anytypeio/any-sync/app"
|
||||
list "github.com/anytypeio/any-sync/commonspace/object/acl/list"
|
||||
objecttree "github.com/anytypeio/any-sync/commonspace/object/tree/objecttree"
|
||||
treestorage "github.com/anytypeio/any-sync/commonspace/object/tree/treestorage"
|
||||
gomock "github.com/golang/mock/gomock"
|
||||
)
|
||||
|
||||
@ -52,20 +50,6 @@ func (mr *MockTreeManagerMockRecorder) Close(arg0 interface{}) *gomock.Call {
|
||||
return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Close", reflect.TypeOf((*MockTreeManager)(nil).Close), arg0)
|
||||
}
|
||||
|
||||
// DeleteSpace mocks base method.
|
||||
func (m *MockTreeManager) DeleteSpace(arg0 context.Context, arg1 string) error {
|
||||
m.ctrl.T.Helper()
|
||||
ret := m.ctrl.Call(m, "DeleteSpace", arg0, arg1)
|
||||
ret0, _ := ret[0].(error)
|
||||
return ret0
|
||||
}
|
||||
|
||||
// DeleteSpace indicates an expected call of DeleteSpace.
|
||||
func (mr *MockTreeManagerMockRecorder) DeleteSpace(arg0, arg1 interface{}) *gomock.Call {
|
||||
mr.mock.ctrl.T.Helper()
|
||||
return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "DeleteSpace", reflect.TypeOf((*MockTreeManager)(nil).DeleteSpace), arg0, arg1)
|
||||
}
|
||||
|
||||
// DeleteTree mocks base method.
|
||||
func (m *MockTreeManager) DeleteTree(arg0 context.Context, arg1, arg2 string) error {
|
||||
m.ctrl.T.Helper()
|
||||
@ -123,20 +107,6 @@ func (mr *MockTreeManagerMockRecorder) Name() *gomock.Call {
|
||||
return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Name", reflect.TypeOf((*MockTreeManager)(nil).Name))
|
||||
}
|
||||
|
||||
// ObjectTreeBuilder mocks base method.
|
||||
func (m *MockTreeManager) ObjectTreeBuilder() func(treestorage.TreeStorage, list.AclList) (objecttree.ObjectTree, error) {
|
||||
m.ctrl.T.Helper()
|
||||
ret := m.ctrl.Call(m, "ObjectTreeBuilder")
|
||||
ret0, _ := ret[0].(func(treestorage.TreeStorage, list.AclList) (objecttree.ObjectTree, error))
|
||||
return ret0
|
||||
}
|
||||
|
||||
// ObjectTreeBuilder indicates an expected call of ObjectTreeBuilder.
|
||||
func (mr *MockTreeManagerMockRecorder) ObjectTreeBuilder() *gomock.Call {
|
||||
mr.mock.ctrl.T.Helper()
|
||||
return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "ObjectTreeBuilder", reflect.TypeOf((*MockTreeManager)(nil).ObjectTreeBuilder))
|
||||
}
|
||||
|
||||
// Run mocks base method.
|
||||
func (m *MockTreeManager) Run(arg0 context.Context) error {
|
||||
m.ctrl.T.Helper()
|
||||
|
||||
@ -13,6 +13,4 @@ type TreeManager interface {
|
||||
app.ComponentRunnable
|
||||
GetTree(ctx context.Context, spaceId, treeId string) (objecttree.ObjectTree, error)
|
||||
DeleteTree(ctx context.Context, spaceId, treeId string) error
|
||||
DeleteSpace(ctx context.Context, spaceId string) error
|
||||
ObjectTreeBuilder() objecttree.BuildObjectTreeFunc
|
||||
}
|
||||
|
||||
138
commonspace/objectsync/mock_objectsync/mock_objectsync.go
Normal file
138
commonspace/objectsync/mock_objectsync/mock_objectsync.go
Normal file
@ -0,0 +1,138 @@
|
||||
// Code generated by MockGen. DO NOT EDIT.
|
||||
// Source: github.com/anytypeio/any-sync/commonspace/objectsync (interfaces: SyncClient)
|
||||
|
||||
// Package mock_objectsync is a generated GoMock package.
|
||||
package mock_objectsync
|
||||
|
||||
import (
|
||||
context "context"
|
||||
reflect "reflect"
|
||||
|
||||
objecttree "github.com/anytypeio/any-sync/commonspace/object/tree/objecttree"
|
||||
treechangeproto "github.com/anytypeio/any-sync/commonspace/object/tree/treechangeproto"
|
||||
peermanager "github.com/anytypeio/any-sync/commonspace/peermanager"
|
||||
gomock "github.com/golang/mock/gomock"
|
||||
)
|
||||
|
||||
// MockSyncClient is a mock of SyncClient interface.
|
||||
type MockSyncClient struct {
|
||||
ctrl *gomock.Controller
|
||||
recorder *MockSyncClientMockRecorder
|
||||
}
|
||||
|
||||
// MockSyncClientMockRecorder is the mock recorder for MockSyncClient.
|
||||
type MockSyncClientMockRecorder struct {
|
||||
mock *MockSyncClient
|
||||
}
|
||||
|
||||
// NewMockSyncClient creates a new mock instance.
|
||||
func NewMockSyncClient(ctrl *gomock.Controller) *MockSyncClient {
|
||||
mock := &MockSyncClient{ctrl: ctrl}
|
||||
mock.recorder = &MockSyncClientMockRecorder{mock}
|
||||
return mock
|
||||
}
|
||||
|
||||
// EXPECT returns an object that allows the caller to indicate expected use.
|
||||
func (m *MockSyncClient) EXPECT() *MockSyncClientMockRecorder {
|
||||
return m.recorder
|
||||
}
|
||||
|
||||
// Broadcast mocks base method.
|
||||
func (m *MockSyncClient) Broadcast(arg0 context.Context, arg1 *treechangeproto.TreeSyncMessage) error {
|
||||
m.ctrl.T.Helper()
|
||||
ret := m.ctrl.Call(m, "Broadcast", arg0, arg1)
|
||||
ret0, _ := ret[0].(error)
|
||||
return ret0
|
||||
}
|
||||
|
||||
// Broadcast indicates an expected call of Broadcast.
|
||||
func (mr *MockSyncClientMockRecorder) Broadcast(arg0, arg1 interface{}) *gomock.Call {
|
||||
mr.mock.ctrl.T.Helper()
|
||||
return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Broadcast", reflect.TypeOf((*MockSyncClient)(nil).Broadcast), arg0, arg1)
|
||||
}
|
||||
|
||||
// CreateFullSyncRequest mocks base method.
|
||||
func (m *MockSyncClient) CreateFullSyncRequest(arg0 objecttree.ObjectTree, arg1, arg2 []string) (*treechangeproto.TreeSyncMessage, error) {
|
||||
m.ctrl.T.Helper()
|
||||
ret := m.ctrl.Call(m, "CreateFullSyncRequest", arg0, arg1, arg2)
|
||||
ret0, _ := ret[0].(*treechangeproto.TreeSyncMessage)
|
||||
ret1, _ := ret[1].(error)
|
||||
return ret0, ret1
|
||||
}
|
||||
|
||||
// CreateFullSyncRequest indicates an expected call of CreateFullSyncRequest.
|
||||
func (mr *MockSyncClientMockRecorder) CreateFullSyncRequest(arg0, arg1, arg2 interface{}) *gomock.Call {
|
||||
mr.mock.ctrl.T.Helper()
|
||||
return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "CreateFullSyncRequest", reflect.TypeOf((*MockSyncClient)(nil).CreateFullSyncRequest), arg0, arg1, arg2)
|
||||
}
|
||||
|
||||
// CreateFullSyncResponse mocks base method.
|
||||
func (m *MockSyncClient) CreateFullSyncResponse(arg0 objecttree.ObjectTree, arg1, arg2 []string) (*treechangeproto.TreeSyncMessage, error) {
|
||||
m.ctrl.T.Helper()
|
||||
ret := m.ctrl.Call(m, "CreateFullSyncResponse", arg0, arg1, arg2)
|
||||
ret0, _ := ret[0].(*treechangeproto.TreeSyncMessage)
|
||||
ret1, _ := ret[1].(error)
|
||||
return ret0, ret1
|
||||
}
|
||||
|
||||
// CreateFullSyncResponse indicates an expected call of CreateFullSyncResponse.
|
||||
func (mr *MockSyncClientMockRecorder) CreateFullSyncResponse(arg0, arg1, arg2 interface{}) *gomock.Call {
|
||||
mr.mock.ctrl.T.Helper()
|
||||
return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "CreateFullSyncResponse", reflect.TypeOf((*MockSyncClient)(nil).CreateFullSyncResponse), arg0, arg1, arg2)
|
||||
}
|
||||
|
||||
// CreateHeadUpdate mocks base method.
|
||||
func (m *MockSyncClient) CreateHeadUpdate(arg0 objecttree.ObjectTree, arg1 []*treechangeproto.RawTreeChangeWithId) *treechangeproto.TreeSyncMessage {
|
||||
m.ctrl.T.Helper()
|
||||
ret := m.ctrl.Call(m, "CreateHeadUpdate", arg0, arg1)
|
||||
ret0, _ := ret[0].(*treechangeproto.TreeSyncMessage)
|
||||
return ret0
|
||||
}
|
||||
|
||||
// CreateHeadUpdate indicates an expected call of CreateHeadUpdate.
|
||||
func (mr *MockSyncClientMockRecorder) CreateHeadUpdate(arg0, arg1 interface{}) *gomock.Call {
|
||||
mr.mock.ctrl.T.Helper()
|
||||
return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "CreateHeadUpdate", reflect.TypeOf((*MockSyncClient)(nil).CreateHeadUpdate), arg0, arg1)
|
||||
}
|
||||
|
||||
// CreateNewTreeRequest mocks base method.
|
||||
func (m *MockSyncClient) CreateNewTreeRequest() *treechangeproto.TreeSyncMessage {
|
||||
m.ctrl.T.Helper()
|
||||
ret := m.ctrl.Call(m, "CreateNewTreeRequest")
|
||||
ret0, _ := ret[0].(*treechangeproto.TreeSyncMessage)
|
||||
return ret0
|
||||
}
|
||||
|
||||
// CreateNewTreeRequest indicates an expected call of CreateNewTreeRequest.
|
||||
func (mr *MockSyncClientMockRecorder) CreateNewTreeRequest() *gomock.Call {
|
||||
mr.mock.ctrl.T.Helper()
|
||||
return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "CreateNewTreeRequest", reflect.TypeOf((*MockSyncClient)(nil).CreateNewTreeRequest))
|
||||
}
|
||||
|
||||
// PeerManager mocks base method.
|
||||
func (m *MockSyncClient) PeerManager() peermanager.PeerManager {
|
||||
m.ctrl.T.Helper()
|
||||
ret := m.ctrl.Call(m, "PeerManager")
|
||||
ret0, _ := ret[0].(peermanager.PeerManager)
|
||||
return ret0
|
||||
}
|
||||
|
||||
// PeerManager indicates an expected call of PeerManager.
|
||||
func (mr *MockSyncClientMockRecorder) PeerManager() *gomock.Call {
|
||||
mr.mock.ctrl.T.Helper()
|
||||
return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "PeerManager", reflect.TypeOf((*MockSyncClient)(nil).PeerManager))
|
||||
}
|
||||
|
||||
// SendWithReply mocks base method.
|
||||
func (m *MockSyncClient) SendWithReply(arg0 context.Context, arg1 string, arg2 *treechangeproto.TreeSyncMessage, arg3 string) error {
|
||||
m.ctrl.T.Helper()
|
||||
ret := m.ctrl.Call(m, "SendWithReply", arg0, arg1, arg2, arg3)
|
||||
ret0, _ := ret[0].(error)
|
||||
return ret0
|
||||
}
|
||||
|
||||
// SendWithReply indicates an expected call of SendWithReply.
|
||||
func (mr *MockSyncClientMockRecorder) SendWithReply(arg0, arg1, arg2, arg3 interface{}) *gomock.Call {
|
||||
mr.mock.ctrl.T.Helper()
|
||||
return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "SendWithReply", reflect.TypeOf((*MockSyncClient)(nil).SendWithReply), arg0, arg1, arg2, arg3)
|
||||
}
|
||||
@ -1,8 +1,8 @@
|
||||
//go:generate mockgen -destination mock_objectsync/mock_objectsync.go github.com/anytypeio/any-sync/commonspace/objectsync SyncClient
|
||||
package objectsync
|
||||
|
||||
import (
|
||||
"context"
|
||||
"github.com/anytypeio/any-sync/commonspace/object/tree/synctree"
|
||||
"github.com/anytypeio/any-sync/commonspace/object/tree/treechangeproto"
|
||||
"github.com/gogo/protobuf/proto"
|
||||
"sync/atomic"
|
||||
@ -33,7 +33,7 @@ type objectSync struct {
|
||||
spaceId string
|
||||
|
||||
messagePool MessagePool
|
||||
syncClient synctree.SyncClient
|
||||
syncClient SyncClient
|
||||
objectGetter syncobjectgetter.SyncObjectGetter
|
||||
configuration nodeconf.Configuration
|
||||
spaceStorage spacestorage.SpaceStorage
|
||||
@ -59,7 +59,7 @@ func NewObjectSync(
|
||||
cancelSync: cancel,
|
||||
spaceIsDeleted: spaceIsDeleted,
|
||||
configuration: configuration,
|
||||
syncClient: synctree.NewSyncClient(spaceId, peerManager, synctree.GetRequestFactory()),
|
||||
syncClient: NewSyncClient(spaceId, peerManager, GetRequestFactory()),
|
||||
}
|
||||
os.messagePool = newMessagePool(peerManager, os.handleMessage)
|
||||
return os
|
||||
@ -90,7 +90,7 @@ func (s *objectSync) handleMessage(ctx context.Context, senderId string, msg *sp
|
||||
log.With(zap.String("objectId", msg.ObjectId), zap.String("replyId", msg.ReplyId)).DebugCtx(ctx, "handling message")
|
||||
obj, err := s.objectGetter.GetObject(ctx, msg.ObjectId)
|
||||
if err != nil {
|
||||
respErr := s.sendErrorResponse(ctx, msg, senderId, err)
|
||||
respErr := s.sendErrorResponse(ctx, msg, senderId)
|
||||
if respErr != nil {
|
||||
log.Debug("failed to send error response", zap.Error(respErr))
|
||||
}
|
||||
@ -103,12 +103,12 @@ func (s *objectSync) MessagePool() MessagePool {
|
||||
return s.messagePool
|
||||
}
|
||||
|
||||
func (s *objectSync) sendErrorResponse(ctx context.Context, msg *spacesyncproto.ObjectSyncMessage, senderId string, respErr error) (err error) {
|
||||
func (s *objectSync) sendErrorResponse(ctx context.Context, msg *spacesyncproto.ObjectSyncMessage, senderId string) (err error) {
|
||||
unmarshalled := &treechangeproto.TreeSyncMessage{}
|
||||
err = proto.Unmarshal(msg.Payload, unmarshalled)
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
resp := treechangeproto.WrapError(respErr, unmarshalled.RootChange)
|
||||
resp := treechangeproto.WrapError(treechangeproto.ErrorCodes_GetTreeError, unmarshalled.RootChange)
|
||||
return s.syncClient.SendWithReply(ctx, senderId, resp, msg.ReplyId)
|
||||
}
|
||||
|
||||
@ -1,4 +1,4 @@
|
||||
package synctree
|
||||
package objectsync
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
@ -1,5 +1,4 @@
|
||||
//go:generate mockgen -destination mock_synctree/mock_synctree.go github.com/anytypeio/any-sync/commonspace/object/tree/synctree SyncClient,SyncTree,ReceiveQueue,HeadNotifiable
|
||||
package synctree
|
||||
package objectsync
|
||||
|
||||
import (
|
||||
"context"
|
||||
@ -12,12 +11,13 @@ type SyncClient interface {
|
||||
RequestFactory
|
||||
Broadcast(ctx context.Context, msg *treechangeproto.TreeSyncMessage) (err error)
|
||||
SendWithReply(ctx context.Context, peerId string, msg *treechangeproto.TreeSyncMessage, replyId string) (err error)
|
||||
PeerManager() peermanager.PeerManager
|
||||
}
|
||||
|
||||
type syncClient struct {
|
||||
peermanager.PeerManager
|
||||
RequestFactory
|
||||
spaceId string
|
||||
spaceId string
|
||||
peerManager peermanager.PeerManager
|
||||
}
|
||||
|
||||
func NewSyncClient(
|
||||
@ -25,29 +25,33 @@ func NewSyncClient(
|
||||
peerManager peermanager.PeerManager,
|
||||
factory RequestFactory) SyncClient {
|
||||
return &syncClient{
|
||||
PeerManager: peerManager,
|
||||
peerManager: peerManager,
|
||||
RequestFactory: factory,
|
||||
spaceId: spaceId,
|
||||
}
|
||||
}
|
||||
|
||||
func (s *syncClient) Broadcast(ctx context.Context, msg *treechangeproto.TreeSyncMessage) (err error) {
|
||||
objMsg, err := marshallTreeMessage(msg, s.spaceId, msg.RootChange.Id, "")
|
||||
objMsg, err := MarshallTreeMessage(msg, s.spaceId, msg.RootChange.Id, "")
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
return s.PeerManager.Broadcast(ctx, objMsg)
|
||||
return s.peerManager.Broadcast(ctx, objMsg)
|
||||
}
|
||||
|
||||
func (s *syncClient) SendWithReply(ctx context.Context, peerId string, msg *treechangeproto.TreeSyncMessage, replyId string) (err error) {
|
||||
objMsg, err := marshallTreeMessage(msg, s.spaceId, msg.RootChange.Id, replyId)
|
||||
objMsg, err := MarshallTreeMessage(msg, s.spaceId, msg.RootChange.Id, replyId)
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
return s.PeerManager.SendPeer(ctx, peerId, objMsg)
|
||||
return s.peerManager.SendPeer(ctx, peerId, objMsg)
|
||||
}
|
||||
|
||||
func marshallTreeMessage(message *treechangeproto.TreeSyncMessage, spaceId, objectId, replyId string) (objMsg *spacesyncproto.ObjectSyncMessage, err error) {
|
||||
func (s *syncClient) PeerManager() peermanager.PeerManager {
|
||||
return s.peerManager
|
||||
}
|
||||
|
||||
func MarshallTreeMessage(message *treechangeproto.TreeSyncMessage, spaceId, objectId, replyId string) (objMsg *spacesyncproto.ObjectSyncMessage, err error) {
|
||||
payload, err := message.Marshal()
|
||||
if err != nil {
|
||||
return
|
||||
@ -55,10 +55,6 @@ func (d *deletionManager) UpdateState(ctx context.Context, state *settingsstate.
|
||||
return nil
|
||||
}
|
||||
log.Debug("deleting space")
|
||||
err = d.treeManager.DeleteSpace(ctx, d.spaceId)
|
||||
if err != nil {
|
||||
log.Debug("failed to notify on space deletion", zap.Error(err))
|
||||
}
|
||||
if d.isResponsible {
|
||||
allIds := slice.DiscardFromSlice(d.provider.AllIds(), func(id string) bool {
|
||||
return id == d.settingsId
|
||||
|
||||
@ -30,7 +30,6 @@ func TestDeletionManager_UpdateState_NotResponsible(t *testing.T) {
|
||||
treeManager := mock_treemanager.NewMockTreeManager(ctrl)
|
||||
|
||||
delState.EXPECT().Add(state.DeletedIds).Return(nil)
|
||||
treeManager.EXPECT().DeleteSpace(ctx, spaceId).Return(nil)
|
||||
|
||||
delManager := newDeletionManager(spaceId,
|
||||
settingsId,
|
||||
@ -64,7 +63,6 @@ func TestDeletionManager_UpdateState_Responsible(t *testing.T) {
|
||||
provider := mock_settings.NewMockSpaceIdsProvider(ctrl)
|
||||
|
||||
delState.EXPECT().Add(state.DeletedIds).Return(nil)
|
||||
treeManager.EXPECT().DeleteSpace(ctx, spaceId).Return(nil)
|
||||
provider.EXPECT().AllIds().Return([]string{"id", "otherId", settingsId})
|
||||
delState.EXPECT().Add([]string{"id", "otherId"}).Return(nil)
|
||||
delManager := newDeletionManager(spaceId,
|
||||
|
||||
@ -124,6 +124,7 @@ type space struct {
|
||||
configuration nodeconf.Configuration
|
||||
settingsObject settings.SettingsObject
|
||||
peerManager peermanager.PeerManager
|
||||
treeBuilder objecttree.BuildObjectTreeFunc
|
||||
|
||||
handleQueue multiqueue.MultiQueue[HandleMessage]
|
||||
|
||||
@ -292,7 +293,7 @@ func (s *space) PutTree(ctx context.Context, payload treestorage.TreeStorageCrea
|
||||
OnClose: s.onObjectClose,
|
||||
SyncStatus: s.syncStatus,
|
||||
PeerGetter: s.peerManager,
|
||||
BuildObjectTree: s.treeManager.ObjectTreeBuilder(),
|
||||
BuildObjectTree: s.treeBuilder,
|
||||
}
|
||||
t, err = synctree.PutSyncTree(ctx, payload, deps)
|
||||
if err != nil {
|
||||
@ -331,7 +332,7 @@ func (s *space) BuildTree(ctx context.Context, id string, opts BuildTreeOpts) (t
|
||||
SyncStatus: s.syncStatus,
|
||||
WaitTreeRemoteSync: opts.WaitTreeRemoteSync,
|
||||
PeerGetter: s.peerManager,
|
||||
BuildObjectTree: s.treeManager.ObjectTreeBuilder(),
|
||||
BuildObjectTree: s.treeBuilder,
|
||||
}
|
||||
if t, err = synctree.BuildSyncTreeOrGetRemote(ctx, id, deps); err != nil {
|
||||
return nil, err
|
||||
|
||||
@ -8,6 +8,7 @@ import (
|
||||
"github.com/anytypeio/any-sync/commonspace/credentialprovider"
|
||||
"github.com/anytypeio/any-sync/commonspace/headsync"
|
||||
"github.com/anytypeio/any-sync/commonspace/object/acl/aclrecordproto"
|
||||
"github.com/anytypeio/any-sync/commonspace/object/tree/objecttree"
|
||||
"github.com/anytypeio/any-sync/commonspace/object/tree/treechangeproto"
|
||||
"github.com/anytypeio/any-sync/commonspace/object/treemanager"
|
||||
"github.com/anytypeio/any-sync/commonspace/objectsync"
|
||||
@ -152,6 +153,12 @@ func (s *spaceService) NewSpace(ctx context.Context, id string) (Space, error) {
|
||||
// TODO: move it to the client package and add possibility to inject StatusProvider from the client
|
||||
syncStatus = syncstatus.NewSyncStatusProvider(st.Id(), syncstatus.DefaultDeps(lastConfiguration, st))
|
||||
}
|
||||
var builder objecttree.BuildObjectTreeFunc
|
||||
if s.config.TreeInMemoryData {
|
||||
builder = objecttree.BuildEmptyDataObjectTree
|
||||
} else {
|
||||
builder = objecttree.BuildObjectTree
|
||||
}
|
||||
|
||||
peerManager, err := s.peermanagerProvider.NewPeerManager(ctx, id)
|
||||
if err != nil {
|
||||
@ -171,6 +178,7 @@ func (s *spaceService) NewSpace(ctx context.Context, id string) (Space, error) {
|
||||
peerManager: peerManager,
|
||||
storage: st,
|
||||
treesUsed: &atomic.Int32{},
|
||||
treeBuilder: builder,
|
||||
isClosed: spaceIsClosed,
|
||||
isDeleted: spaceIsDeleted,
|
||||
}
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user