diff --git a/Makefile b/Makefile index aa0fa75f..003fbbf7 100644 --- a/Makefile +++ b/Makefile @@ -20,6 +20,7 @@ proto: protoc --gogofaster_out=$(PKGMAP):. --go-drpc_out=protolib=github.com/gogo/protobuf:. net/streampool/testservice/protos/*.proto protoc --gogofaster_out=:. net/secureservice/handshake/handshakeproto/protos/*.proto protoc --gogofaster_out=$(PKGMAP):. --go-drpc_out=protolib=github.com/gogo/protobuf:. coordinator/coordinatorproto/protos/*.proto + protoc --gogofaster_out=:. --go-drpc_out=protolib=github.com/gogo/protobuf:. consensus/consensusproto/protos/*.proto deps: go mod download diff --git a/consensus/consensusclient/client.go b/consensus/consensusclient/client.go new file mode 100644 index 00000000..cd3c411a --- /dev/null +++ b/consensus/consensusclient/client.go @@ -0,0 +1,240 @@ +//go:generate mockgen -destination mock_consensusclient/mock_consensusclient.go github.com/anyproto/any-sync/consensus/consensusclient Service +package consensusclient + +import ( + "context" + "errors" + "github.com/anyproto/any-sync/app" + "github.com/anyproto/any-sync/app/logger" + "github.com/anyproto/any-sync/consensus/consensusproto" + "github.com/anyproto/any-sync/net/pool" + "github.com/anyproto/any-sync/net/rpc/rpcerr" + "github.com/anyproto/any-sync/nodeconf" + "go.uber.org/zap" + "sync" + "time" +) + +const CName = "consensus.consensusclient" + +var log = logger.NewNamed(CName) + +var ( + ErrWatcherExists = errors.New("watcher exists") + ErrWatcherNotExists = errors.New("watcher not exists") +) + +func New() Service { + return new(service) +} + +// Watcher watches new events by specified logId +type Watcher interface { + AddConsensusRecords(recs []*consensusproto.Record) + AddConsensusError(err error) +} + +type Service interface { + // AddLog adds new log to consensus servers + AddLog(ctx context.Context, clog *consensusproto.Log) (err error) + // AddRecord adds new record to consensus servers + AddRecord(ctx context.Context, logId []byte, clog *consensusproto.Record) (err error) + // Watch starts watching to given logId and calls watcher when any relative event received + Watch(logId []byte, w Watcher) (err error) + // UnWatch stops watching given logId and removes watcher + UnWatch(logId []byte) (err error) + app.ComponentRunnable +} + +type service struct { + pool pool.Pool + nodeconf nodeconf.Service + + watchers map[string]Watcher + stream *stream + close chan struct{} + mu sync.Mutex +} + +func (s *service) Init(a *app.App) (err error) { + s.pool = a.MustComponent(pool.CName).(pool.Pool) + s.nodeconf = a.MustComponent(nodeconf.CName).(nodeconf.Service) + s.watchers = make(map[string]Watcher) + s.close = make(chan struct{}) + return nil +} + +func (s *service) Name() (name string) { + return CName +} + +func (s *service) Run(_ context.Context) error { + go s.streamWatcher() + return nil +} + +func (s *service) doClient(ctx context.Context, fn func(cl consensusproto.DRPCConsensusClient) error) error { + peer, err := s.pool.GetOneOf(ctx, s.nodeconf.ConsensusPeers()) + if err != nil { + return err + } + dc, err := peer.AcquireDrpcConn(ctx) + if err != nil { + return err + } + defer peer.ReleaseDrpcConn(dc) + return fn(consensusproto.NewDRPCConsensusClient(dc)) +} + +func (s *service) AddLog(ctx context.Context, clog *consensusproto.Log) (err error) { + return s.doClient(ctx, func(cl consensusproto.DRPCConsensusClient) error { + if _, err = cl.LogAdd(ctx, &consensusproto.LogAddRequest{ + Log: clog, + }); err != nil { + return rpcerr.Unwrap(err) + } + return nil + }) +} + +func (s *service) AddRecord(ctx context.Context, logId []byte, clog *consensusproto.Record) (err error) { + return s.doClient(ctx, func(cl consensusproto.DRPCConsensusClient) error { + if _, err = cl.RecordAdd(ctx, &consensusproto.RecordAddRequest{ + LogId: logId, + Record: clog, + }); err != nil { + return rpcerr.Unwrap(err) + } + return nil + }) +} + +func (s *service) Watch(logId []byte, w Watcher) (err error) { + s.mu.Lock() + defer s.mu.Unlock() + if _, ok := s.watchers[string(logId)]; ok { + return ErrWatcherExists + } + s.watchers[string(logId)] = w + if s.stream != nil { + if wErr := s.stream.WatchIds([][]byte{logId}); wErr != nil { + log.Warn("WatchIds error", zap.Error(wErr)) + } + } + return +} + +func (s *service) UnWatch(logId []byte) (err error) { + s.mu.Lock() + defer s.mu.Unlock() + if _, ok := s.watchers[string(logId)]; !ok { + return ErrWatcherNotExists + } + delete(s.watchers, string(logId)) + if s.stream != nil { + if wErr := s.stream.UnwatchIds([][]byte{logId}); wErr != nil { + log.Warn("UnWatchIds error", zap.Error(wErr)) + } + } + return +} + +func (s *service) openStream(ctx context.Context) (st *stream, err error) { + pr, err := s.pool.GetOneOf(ctx, s.nodeconf.ConsensusPeers()) + if err != nil { + return nil, err + } + dc, err := pr.AcquireDrpcConn(ctx) + if err != nil { + return nil, err + } + rpcStream, err := consensusproto.NewDRPCConsensusClient(dc).LogWatch(ctx) + if err != nil { + return nil, rpcerr.Unwrap(err) + } + return runStream(rpcStream), nil +} + +func (s *service) streamWatcher() { + var ( + err error + st *stream + i int + ) + for { + // open stream + if st, err = s.openStream(context.Background()); err != nil { + // can't open stream, we will retry until success connection or close + if i < 60 { + i++ + } + sleepTime := time.Second * time.Duration(i) + log.Error("watch log error", zap.Error(err), zap.Duration("waitTime", sleepTime)) + select { + case <-time.After(sleepTime): + continue + case <-s.close: + return + } + } + i = 0 + + // collect ids and setup stream + s.mu.Lock() + var logIds = make([][]byte, 0, len(s.watchers)) + for id := range s.watchers { + logIds = append(logIds, []byte(id)) + } + s.stream = st + s.mu.Unlock() + + // restore subscriptions + if len(logIds) > 0 { + if err = s.stream.WatchIds(logIds); err != nil { + log.Error("watch ids error", zap.Error(err)) + continue + } + } + + // read stream + if err = s.streamReader(); err != nil { + log.Error("stream read error", zap.Error(err)) + continue + } + return + } +} + +func (s *service) streamReader() error { + for { + events := s.stream.WaitLogs() + if len(events) == 0 { + return s.stream.Err() + } + for _, e := range events { + if w, ok := s.watchers[string(e.LogId)]; ok { + if e.Error == nil { + w.AddConsensusRecords(e.Records) + } else { + w.AddConsensusError(rpcerr.Err(uint64(e.Error.Error))) + } + } else { + log.Warn("received unexpected log id", zap.Binary("logId", e.LogId)) + } + } + } +} + +func (s *service) Close(_ context.Context) error { + s.mu.Lock() + if s.stream != nil { + _ = s.stream.Close() + } + s.mu.Unlock() + select { + case <-s.close: + default: + close(s.close) + } + return nil +} diff --git a/consensus/consensusclient/client_test.go b/consensus/consensusclient/client_test.go new file mode 100644 index 00000000..a212a635 --- /dev/null +++ b/consensus/consensusclient/client_test.go @@ -0,0 +1,236 @@ +package consensusclient + +import ( + "context" + "fmt" + "github.com/anyproto/any-sync/app" + "github.com/anyproto/any-sync/consensus/consensusproto" + "github.com/anyproto/any-sync/consensus/consensusproto/consensuserr" + "github.com/anyproto/any-sync/net/pool" + "github.com/anyproto/any-sync/net/rpc/rpctest" + "github.com/anyproto/any-sync/nodeconf" + "github.com/anyproto/any-sync/nodeconf/mock_nodeconf" + "github.com/anyproto/any-sync/testutil/accounttest" + "github.com/golang/mock/gomock" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + "sync" + "testing" + "time" +) + +func TestService_Watch(t *testing.T) { + t.Run("not found error", func(t *testing.T) { + fx := newFixture(t).run(t) + defer fx.Finish() + var logId = []byte{'1'} + w := &testWatcher{ready: make(chan struct{})} + require.NoError(t, fx.Watch(logId, w)) + st := fx.testServer.waitStream(t) + req, err := st.Recv() + require.NoError(t, err) + assert.Equal(t, [][]byte{logId}, req.WatchIds) + require.NoError(t, st.Send(&consensusproto.LogWatchEvent{ + LogId: logId, + Error: &consensusproto.Err{ + Error: consensusproto.ErrCodes_ErrorOffset + consensusproto.ErrCodes_LogNotFound, + }, + })) + <-w.ready + assert.Equal(t, consensuserr.ErrLogNotFound, w.err) + fx.testServer.releaseStream <- nil + }) + t.Run("watcherExists error", func(t *testing.T) { + fx := newFixture(t).run(t) + defer fx.Finish() + var logId = []byte{'1'} + w := &testWatcher{} + require.NoError(t, fx.Watch(logId, w)) + require.Error(t, fx.Watch(logId, w)) + st := fx.testServer.waitStream(t) + st.Recv() + fx.testServer.releaseStream <- nil + }) + t.Run("watch", func(t *testing.T) { + fx := newFixture(t).run(t) + defer fx.Finish() + var logId1 = []byte{'1'} + w := &testWatcher{} + require.NoError(t, fx.Watch(logId1, w)) + st := fx.testServer.waitStream(t) + req, err := st.Recv() + require.NoError(t, err) + assert.Equal(t, [][]byte{logId1}, req.WatchIds) + + var logId2 = []byte{'2'} + w = &testWatcher{} + require.NoError(t, fx.Watch(logId2, w)) + req, err = st.Recv() + require.NoError(t, err) + assert.Equal(t, [][]byte{logId2}, req.WatchIds) + + fx.testServer.releaseStream <- nil + }) +} + +func TestService_UnWatch(t *testing.T) { + t.Run("no watcher", func(t *testing.T) { + fx := newFixture(t).run(t) + defer fx.Finish() + require.Error(t, fx.UnWatch([]byte{'1'})) + }) + t.Run("success", func(t *testing.T) { + fx := newFixture(t).run(t) + defer fx.Finish() + w := &testWatcher{} + require.NoError(t, fx.Watch([]byte{'1'}, w)) + assert.NoError(t, fx.UnWatch([]byte{'1'})) + }) +} + +func TestService_Init(t *testing.T) { + t.Run("reconnect on watch err", func(t *testing.T) { + fx := newFixture(t) + fx.testServer.watchErrOnce = true + fx.run(t) + defer fx.Finish() + fx.testServer.waitStream(t) + fx.testServer.releaseStream <- nil + }) + t.Run("reconnect on start", func(t *testing.T) { + fx := newFixture(t) + fx.a.MustComponent(pool.CName).(*rpctest.TestPool).WithServer(nil) + fx.run(t) + defer fx.Finish() + time.Sleep(time.Millisecond * 50) + fx.a.MustComponent(pool.CName).(*rpctest.TestPool).WithServer(fx.drpcTS) + fx.testServer.waitStream(t) + fx.testServer.releaseStream <- nil + }) +} + +func TestService_AddLog(t *testing.T) { + fx := newFixture(t).run(t) + defer fx.Finish() + assert.NoError(t, fx.AddLog(ctx, &consensusproto.Log{})) +} + +func TestService_AddRecord(t *testing.T) { + fx := newFixture(t).run(t) + defer fx.Finish() + assert.NoError(t, fx.AddRecord(ctx, []byte{'1'}, &consensusproto.Record{})) +} + +var ctx = context.Background() + +func newFixture(t *testing.T) *fixture { + fx := &fixture{ + Service: New(), + a: &app.App{}, + ctrl: gomock.NewController(t), + testServer: &testServer{ + stream: make(chan consensusproto.DRPCConsensus_LogWatchStream), + releaseStream: make(chan error), + }, + } + fx.nodeconf = mock_nodeconf.NewMockService(fx.ctrl) + fx.nodeconf.EXPECT().Name().Return(nodeconf.CName).AnyTimes() + fx.nodeconf.EXPECT().Init(gomock.Any()).AnyTimes() + fx.nodeconf.EXPECT().Run(gomock.Any()).AnyTimes() + fx.nodeconf.EXPECT().Close(gomock.Any()).AnyTimes() + fx.nodeconf.EXPECT().ConsensusPeers().Return([]string{"c1", "c2", "c3"}).AnyTimes() + + fx.drpcTS = rpctest.NewTestServer() + require.NoError(t, consensusproto.DRPCRegisterConsensus(fx.drpcTS.Mux, fx.testServer)) + fx.a.Register(fx.Service). + Register(&accounttest.AccountTestService{}). + Register(fx.nodeconf). + Register(rpctest.NewTestPool().WithServer(fx.drpcTS)) + + return fx +} + +type fixture struct { + Service + a *app.App + ctrl *gomock.Controller + testServer *testServer + drpcTS *rpctest.TestServer + nodeconf *mock_nodeconf.MockService +} + +func (fx *fixture) run(t *testing.T) *fixture { + require.NoError(t, fx.a.Start(ctx)) + return fx +} + +func (fx *fixture) Finish() { + assert.NoError(fx.ctrl.T, fx.a.Close(ctx)) + fx.ctrl.Finish() +} + +type testServer struct { + stream chan consensusproto.DRPCConsensus_LogWatchStream + addLog func(ctx context.Context, req *consensusproto.LogAddRequest) error + addRecord func(ctx context.Context, req *consensusproto.RecordAddRequest) error + releaseStream chan error + watchErrOnce bool +} + +func (t *testServer) LogAdd(ctx context.Context, req *consensusproto.LogAddRequest) (*consensusproto.Ok, error) { + if t.addLog != nil { + if err := t.addLog(ctx, req); err != nil { + return nil, err + } + } + return &consensusproto.Ok{}, nil +} + +func (t *testServer) RecordAdd(ctx context.Context, req *consensusproto.RecordAddRequest) (*consensusproto.Ok, error) { + if t.addRecord != nil { + if err := t.addRecord(ctx, req); err != nil { + return nil, err + } + } + return &consensusproto.Ok{}, nil +} + +func (t *testServer) LogWatch(stream consensusproto.DRPCConsensus_LogWatchStream) error { + if t.watchErrOnce { + t.watchErrOnce = false + return fmt.Errorf("error") + } + t.stream <- stream + return <-t.releaseStream +} + +func (t *testServer) waitStream(test *testing.T) consensusproto.DRPCConsensus_LogWatchStream { + select { + case <-time.After(time.Second * 5): + test.Fatalf("waiteStream timeout") + case st := <-t.stream: + return st + } + return nil +} + +type testWatcher struct { + recs [][]*consensusproto.Record + err error + ready chan struct{} + once sync.Once +} + +func (t *testWatcher) AddConsensusRecords(recs []*consensusproto.Record) { + t.recs = append(t.recs, recs) + t.once.Do(func() { + close(t.ready) + }) +} + +func (t *testWatcher) AddConsensusError(err error) { + t.err = err + t.once.Do(func() { + close(t.ready) + }) +} diff --git a/consensus/consensusclient/mock_consensusclient/mock_consensusclient.go b/consensus/consensusclient/mock_consensusclient/mock_consensusclient.go new file mode 100644 index 00000000..d103635a --- /dev/null +++ b/consensus/consensusclient/mock_consensusclient/mock_consensusclient.go @@ -0,0 +1,150 @@ +// Code generated by MockGen. DO NOT EDIT. +// Source: github.com/anyproto/go-anytype-infrastructure-experiments/consensus/consensusclient (interfaces: Service) + +// Package mock_consensusclient is a generated GoMock package. +package mock_consensusclient + +import ( + context "context" + reflect "reflect" + + app "github.com/anyproto/any-sync/app" + consensusclient "github.com/anyproto/any-sync-consensusnode/consensusclient" + consensusproto "github.com/anyproto/any-sync-consensusnode/consensusproto" + gomock "github.com/golang/mock/gomock" +) + +// MockService is a mock of Service interface. +type MockService struct { + ctrl *gomock.Controller + recorder *MockServiceMockRecorder +} + +// MockServiceMockRecorder is the mock recorder for MockService. +type MockServiceMockRecorder struct { + mock *MockService +} + +// NewMockService creates a new mock instance. +func NewMockService(ctrl *gomock.Controller) *MockService { + mock := &MockService{ctrl: ctrl} + mock.recorder = &MockServiceMockRecorder{mock} + return mock +} + +// EXPECT returns an object that allows the caller to indicate expected use. +func (m *MockService) EXPECT() *MockServiceMockRecorder { + return m.recorder +} + +// AddLog mocks base method. +func (m *MockService) AddLog(arg0 context.Context, arg1 *consensusproto.Log) error { + m.ctrl.T.Helper() + ret := m.ctrl.Call(m, "AddLog", arg0, arg1) + ret0, _ := ret[0].(error) + return ret0 +} + +// AddLog indicates an expected call of AddLog. +func (mr *MockServiceMockRecorder) AddLog(arg0, arg1 interface{}) *gomock.Call { + mr.mock.ctrl.T.Helper() + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "AddLog", reflect.TypeOf((*MockService)(nil).AddLog), arg0, arg1) +} + +// AddRecord mocks base method. +func (m *MockService) AddRecord(arg0 context.Context, arg1 []byte, arg2 *consensusproto.Record) error { + m.ctrl.T.Helper() + ret := m.ctrl.Call(m, "AddRecord", arg0, arg1, arg2) + ret0, _ := ret[0].(error) + return ret0 +} + +// AddRecord indicates an expected call of AddRecord. +func (mr *MockServiceMockRecorder) AddRecord(arg0, arg1, arg2 interface{}) *gomock.Call { + mr.mock.ctrl.T.Helper() + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "AddRecord", reflect.TypeOf((*MockService)(nil).AddRecord), arg0, arg1, arg2) +} + +// Close mocks base method. +func (m *MockService) Close(arg0 context.Context) error { + m.ctrl.T.Helper() + ret := m.ctrl.Call(m, "Close", arg0) + ret0, _ := ret[0].(error) + return ret0 +} + +// Close indicates an expected call of Close. +func (mr *MockServiceMockRecorder) Close(arg0 interface{}) *gomock.Call { + mr.mock.ctrl.T.Helper() + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Close", reflect.TypeOf((*MockService)(nil).Close), arg0) +} + +// Init mocks base method. +func (m *MockService) Init(arg0 *app.App) error { + m.ctrl.T.Helper() + ret := m.ctrl.Call(m, "Init", arg0) + ret0, _ := ret[0].(error) + return ret0 +} + +// Init indicates an expected call of Init. +func (mr *MockServiceMockRecorder) Init(arg0 interface{}) *gomock.Call { + mr.mock.ctrl.T.Helper() + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Init", reflect.TypeOf((*MockService)(nil).Init), arg0) +} + +// Name mocks base method. +func (m *MockService) Name() string { + m.ctrl.T.Helper() + ret := m.ctrl.Call(m, "Name") + ret0, _ := ret[0].(string) + return ret0 +} + +// Name indicates an expected call of Name. +func (mr *MockServiceMockRecorder) Name() *gomock.Call { + mr.mock.ctrl.T.Helper() + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Name", reflect.TypeOf((*MockService)(nil).Name)) +} + +// Run mocks base method. +func (m *MockService) Run(arg0 context.Context) error { + m.ctrl.T.Helper() + ret := m.ctrl.Call(m, "Run", arg0) + ret0, _ := ret[0].(error) + return ret0 +} + +// Run indicates an expected call of Run. +func (mr *MockServiceMockRecorder) Run(arg0 interface{}) *gomock.Call { + mr.mock.ctrl.T.Helper() + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Run", reflect.TypeOf((*MockService)(nil).Run), arg0) +} + +// UnWatch mocks base method. +func (m *MockService) UnWatch(arg0 []byte) error { + m.ctrl.T.Helper() + ret := m.ctrl.Call(m, "UnWatch", arg0) + ret0, _ := ret[0].(error) + return ret0 +} + +// UnWatch indicates an expected call of UnWatch. +func (mr *MockServiceMockRecorder) UnWatch(arg0 interface{}) *gomock.Call { + mr.mock.ctrl.T.Helper() + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "UnWatch", reflect.TypeOf((*MockService)(nil).UnWatch), arg0) +} + +// Watch mocks base method. +func (m *MockService) Watch(arg0 []byte, arg1 consensusclient.Watcher) error { + m.ctrl.T.Helper() + ret := m.ctrl.Call(m, "Watch", arg0, arg1) + ret0, _ := ret[0].(error) + return ret0 +} + +// Watch indicates an expected call of Watch. +func (mr *MockServiceMockRecorder) Watch(arg0, arg1 interface{}) *gomock.Call { + mr.mock.ctrl.T.Helper() + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Watch", reflect.TypeOf((*MockService)(nil).Watch), arg0, arg1) +} diff --git a/consensus/consensusclient/stream.go b/consensus/consensusclient/stream.go new file mode 100644 index 00000000..679dcd74 --- /dev/null +++ b/consensus/consensusclient/stream.go @@ -0,0 +1,70 @@ +package consensusclient + +import ( + "context" + "github.com/anyproto/any-sync/consensus/consensusproto" + "github.com/cheggaaa/mb/v3" + "sync" +) + +func runStream(rpcStream consensusproto.DRPCConsensus_LogWatchClient) *stream { + st := &stream{ + rpcStream: rpcStream, + mb: mb.New[*consensusproto.LogWatchEvent](100), + } + go st.readStream() + return st +} + +type stream struct { + rpcStream consensusproto.DRPCConsensus_LogWatchClient + mb *mb.MB[*consensusproto.LogWatchEvent] + mu sync.Mutex + err error +} + +func (s *stream) WatchIds(logIds [][]byte) (err error) { + return s.rpcStream.Send(&consensusproto.LogWatchRequest{ + WatchIds: logIds, + }) +} + +func (s *stream) UnwatchIds(logIds [][]byte) (err error) { + return s.rpcStream.Send(&consensusproto.LogWatchRequest{ + UnwatchIds: logIds, + }) +} + +func (s *stream) WaitLogs() []*consensusproto.LogWatchEvent { + events, _ := s.mb.Wait(context.TODO()) + return events +} + +func (s *stream) Err() error { + s.mu.Lock() + defer s.mu.Unlock() + return s.err +} + +func (s *stream) readStream() { + defer s.Close() + for { + event, err := s.rpcStream.Recv() + if err != nil { + s.mu.Lock() + s.err = err + s.mu.Unlock() + return + } + if err = s.mb.Add(s.rpcStream.Context(), event); err != nil { + return + } + } +} + +func (s *stream) Close() error { + if err := s.mb.Close(); err == nil { + return s.rpcStream.Close() + } + return nil +} diff --git a/consensus/consensusproto/consensus.pb.go b/consensus/consensusproto/consensus.pb.go new file mode 100644 index 00000000..9cb94880 --- /dev/null +++ b/consensus/consensusproto/consensus.pb.go @@ -0,0 +1,1957 @@ +// Code generated by protoc-gen-gogo. DO NOT EDIT. +// source: consensus/consensusproto/protos/consensus.proto + +package consensusproto + +import ( + fmt "fmt" + proto "github.com/gogo/protobuf/proto" + io "io" + math "math" + math_bits "math/bits" +) + +// Reference imports to suppress errors if they are not otherwise used. +var _ = proto.Marshal +var _ = fmt.Errorf +var _ = math.Inf + +// This is a compile-time assertion to ensure that this generated file +// is compatible with the proto package it is being compiled against. +// A compilation error at this line likely means your copy of the +// proto package needs to be updated. +const _ = proto.GoGoProtoPackageIsVersion3 // please upgrade the proto package + +type ErrCodes int32 + +const ( + ErrCodes_Unexpected ErrCodes = 0 + ErrCodes_LogExists ErrCodes = 1 + ErrCodes_LogNotFound ErrCodes = 2 + ErrCodes_RecordConflict ErrCodes = 3 + ErrCodes_ErrorOffset ErrCodes = 300 +) + +var ErrCodes_name = map[int32]string{ + 0: "Unexpected", + 1: "LogExists", + 2: "LogNotFound", + 3: "RecordConflict", + 300: "ErrorOffset", +} + +var ErrCodes_value = map[string]int32{ + "Unexpected": 0, + "LogExists": 1, + "LogNotFound": 2, + "RecordConflict": 3, + "ErrorOffset": 300, +} + +func (x ErrCodes) String() string { + return proto.EnumName(ErrCodes_name, int32(x)) +} + +func (ErrCodes) EnumDescriptor() ([]byte, []int) { + return fileDescriptor_b8d7f1c16b400059, []int{0} +} + +type Log struct { + Id []byte `protobuf:"bytes,1,opt,name=id,proto3" json:"id,omitempty"` + Records []*Record `protobuf:"bytes,2,rep,name=records,proto3" json:"records,omitempty"` +} + +func (m *Log) Reset() { *m = Log{} } +func (m *Log) String() string { return proto.CompactTextString(m) } +func (*Log) ProtoMessage() {} +func (*Log) Descriptor() ([]byte, []int) { + return fileDescriptor_b8d7f1c16b400059, []int{0} +} +func (m *Log) XXX_Unmarshal(b []byte) error { + return m.Unmarshal(b) +} +func (m *Log) XXX_Marshal(b []byte, deterministic bool) ([]byte, error) { + if deterministic { + return xxx_messageInfo_Log.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 *Log) XXX_Merge(src proto.Message) { + xxx_messageInfo_Log.Merge(m, src) +} +func (m *Log) XXX_Size() int { + return m.Size() +} +func (m *Log) XXX_DiscardUnknown() { + xxx_messageInfo_Log.DiscardUnknown(m) +} + +var xxx_messageInfo_Log proto.InternalMessageInfo + +func (m *Log) GetId() []byte { + if m != nil { + return m.Id + } + return nil +} + +func (m *Log) GetRecords() []*Record { + if m != nil { + return m.Records + } + return nil +} + +type Record struct { + Id []byte `protobuf:"bytes,1,opt,name=id,proto3" json:"id,omitempty"` + PrevId []byte `protobuf:"bytes,2,opt,name=prevId,proto3" json:"prevId,omitempty"` + Payload []byte `protobuf:"bytes,3,opt,name=payload,proto3" json:"payload,omitempty"` + CreatedUnix uint64 `protobuf:"varint,4,opt,name=createdUnix,proto3" json:"createdUnix,omitempty"` +} + +func (m *Record) Reset() { *m = Record{} } +func (m *Record) String() string { return proto.CompactTextString(m) } +func (*Record) ProtoMessage() {} +func (*Record) Descriptor() ([]byte, []int) { + return fileDescriptor_b8d7f1c16b400059, []int{1} +} +func (m *Record) XXX_Unmarshal(b []byte) error { + return m.Unmarshal(b) +} +func (m *Record) XXX_Marshal(b []byte, deterministic bool) ([]byte, error) { + if deterministic { + return xxx_messageInfo_Record.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 *Record) XXX_Merge(src proto.Message) { + xxx_messageInfo_Record.Merge(m, src) +} +func (m *Record) XXX_Size() int { + return m.Size() +} +func (m *Record) XXX_DiscardUnknown() { + xxx_messageInfo_Record.DiscardUnknown(m) +} + +var xxx_messageInfo_Record proto.InternalMessageInfo + +func (m *Record) GetId() []byte { + if m != nil { + return m.Id + } + return nil +} + +func (m *Record) GetPrevId() []byte { + if m != nil { + return m.PrevId + } + return nil +} + +func (m *Record) GetPayload() []byte { + if m != nil { + return m.Payload + } + return nil +} + +func (m *Record) GetCreatedUnix() uint64 { + if m != nil { + return m.CreatedUnix + } + return 0 +} + +type Ok struct { +} + +func (m *Ok) Reset() { *m = Ok{} } +func (m *Ok) String() string { return proto.CompactTextString(m) } +func (*Ok) ProtoMessage() {} +func (*Ok) Descriptor() ([]byte, []int) { + return fileDescriptor_b8d7f1c16b400059, []int{2} +} +func (m *Ok) XXX_Unmarshal(b []byte) error { + return m.Unmarshal(b) +} +func (m *Ok) XXX_Marshal(b []byte, deterministic bool) ([]byte, error) { + if deterministic { + return xxx_messageInfo_Ok.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 *Ok) XXX_Merge(src proto.Message) { + xxx_messageInfo_Ok.Merge(m, src) +} +func (m *Ok) XXX_Size() int { + return m.Size() +} +func (m *Ok) XXX_DiscardUnknown() { + xxx_messageInfo_Ok.DiscardUnknown(m) +} + +var xxx_messageInfo_Ok proto.InternalMessageInfo + +type LogAddRequest struct { + Log *Log `protobuf:"bytes,1,opt,name=log,proto3" json:"log,omitempty"` +} + +func (m *LogAddRequest) Reset() { *m = LogAddRequest{} } +func (m *LogAddRequest) String() string { return proto.CompactTextString(m) } +func (*LogAddRequest) ProtoMessage() {} +func (*LogAddRequest) Descriptor() ([]byte, []int) { + return fileDescriptor_b8d7f1c16b400059, []int{3} +} +func (m *LogAddRequest) XXX_Unmarshal(b []byte) error { + return m.Unmarshal(b) +} +func (m *LogAddRequest) XXX_Marshal(b []byte, deterministic bool) ([]byte, error) { + if deterministic { + return xxx_messageInfo_LogAddRequest.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 *LogAddRequest) XXX_Merge(src proto.Message) { + xxx_messageInfo_LogAddRequest.Merge(m, src) +} +func (m *LogAddRequest) XXX_Size() int { + return m.Size() +} +func (m *LogAddRequest) XXX_DiscardUnknown() { + xxx_messageInfo_LogAddRequest.DiscardUnknown(m) +} + +var xxx_messageInfo_LogAddRequest proto.InternalMessageInfo + +func (m *LogAddRequest) GetLog() *Log { + if m != nil { + return m.Log + } + return nil +} + +type RecordAddRequest struct { + LogId []byte `protobuf:"bytes,1,opt,name=logId,proto3" json:"logId,omitempty"` + Record *Record `protobuf:"bytes,2,opt,name=record,proto3" json:"record,omitempty"` +} + +func (m *RecordAddRequest) Reset() { *m = RecordAddRequest{} } +func (m *RecordAddRequest) String() string { return proto.CompactTextString(m) } +func (*RecordAddRequest) ProtoMessage() {} +func (*RecordAddRequest) Descriptor() ([]byte, []int) { + return fileDescriptor_b8d7f1c16b400059, []int{4} +} +func (m *RecordAddRequest) XXX_Unmarshal(b []byte) error { + return m.Unmarshal(b) +} +func (m *RecordAddRequest) XXX_Marshal(b []byte, deterministic bool) ([]byte, error) { + if deterministic { + return xxx_messageInfo_RecordAddRequest.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 *RecordAddRequest) XXX_Merge(src proto.Message) { + xxx_messageInfo_RecordAddRequest.Merge(m, src) +} +func (m *RecordAddRequest) XXX_Size() int { + return m.Size() +} +func (m *RecordAddRequest) XXX_DiscardUnknown() { + xxx_messageInfo_RecordAddRequest.DiscardUnknown(m) +} + +var xxx_messageInfo_RecordAddRequest proto.InternalMessageInfo + +func (m *RecordAddRequest) GetLogId() []byte { + if m != nil { + return m.LogId + } + return nil +} + +func (m *RecordAddRequest) GetRecord() *Record { + if m != nil { + return m.Record + } + return nil +} + +type LogWatchRequest struct { + WatchIds [][]byte `protobuf:"bytes,1,rep,name=watchIds,proto3" json:"watchIds,omitempty"` + UnwatchIds [][]byte `protobuf:"bytes,2,rep,name=unwatchIds,proto3" json:"unwatchIds,omitempty"` +} + +func (m *LogWatchRequest) Reset() { *m = LogWatchRequest{} } +func (m *LogWatchRequest) String() string { return proto.CompactTextString(m) } +func (*LogWatchRequest) ProtoMessage() {} +func (*LogWatchRequest) Descriptor() ([]byte, []int) { + return fileDescriptor_b8d7f1c16b400059, []int{5} +} +func (m *LogWatchRequest) XXX_Unmarshal(b []byte) error { + return m.Unmarshal(b) +} +func (m *LogWatchRequest) XXX_Marshal(b []byte, deterministic bool) ([]byte, error) { + if deterministic { + return xxx_messageInfo_LogWatchRequest.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 *LogWatchRequest) XXX_Merge(src proto.Message) { + xxx_messageInfo_LogWatchRequest.Merge(m, src) +} +func (m *LogWatchRequest) XXX_Size() int { + return m.Size() +} +func (m *LogWatchRequest) XXX_DiscardUnknown() { + xxx_messageInfo_LogWatchRequest.DiscardUnknown(m) +} + +var xxx_messageInfo_LogWatchRequest proto.InternalMessageInfo + +func (m *LogWatchRequest) GetWatchIds() [][]byte { + if m != nil { + return m.WatchIds + } + return nil +} + +func (m *LogWatchRequest) GetUnwatchIds() [][]byte { + if m != nil { + return m.UnwatchIds + } + return nil +} + +type LogWatchEvent struct { + LogId []byte `protobuf:"bytes,1,opt,name=logId,proto3" json:"logId,omitempty"` + Records []*Record `protobuf:"bytes,2,rep,name=records,proto3" json:"records,omitempty"` + Error *Err `protobuf:"bytes,3,opt,name=error,proto3" json:"error,omitempty"` +} + +func (m *LogWatchEvent) Reset() { *m = LogWatchEvent{} } +func (m *LogWatchEvent) String() string { return proto.CompactTextString(m) } +func (*LogWatchEvent) ProtoMessage() {} +func (*LogWatchEvent) Descriptor() ([]byte, []int) { + return fileDescriptor_b8d7f1c16b400059, []int{6} +} +func (m *LogWatchEvent) XXX_Unmarshal(b []byte) error { + return m.Unmarshal(b) +} +func (m *LogWatchEvent) XXX_Marshal(b []byte, deterministic bool) ([]byte, error) { + if deterministic { + return xxx_messageInfo_LogWatchEvent.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 *LogWatchEvent) XXX_Merge(src proto.Message) { + xxx_messageInfo_LogWatchEvent.Merge(m, src) +} +func (m *LogWatchEvent) XXX_Size() int { + return m.Size() +} +func (m *LogWatchEvent) XXX_DiscardUnknown() { + xxx_messageInfo_LogWatchEvent.DiscardUnknown(m) +} + +var xxx_messageInfo_LogWatchEvent proto.InternalMessageInfo + +func (m *LogWatchEvent) GetLogId() []byte { + if m != nil { + return m.LogId + } + return nil +} + +func (m *LogWatchEvent) GetRecords() []*Record { + if m != nil { + return m.Records + } + return nil +} + +func (m *LogWatchEvent) GetError() *Err { + if m != nil { + return m.Error + } + return nil +} + +type Err struct { + Error ErrCodes `protobuf:"varint,1,opt,name=error,proto3,enum=consensusProto.ErrCodes" json:"error,omitempty"` +} + +func (m *Err) Reset() { *m = Err{} } +func (m *Err) String() string { return proto.CompactTextString(m) } +func (*Err) ProtoMessage() {} +func (*Err) Descriptor() ([]byte, []int) { + return fileDescriptor_b8d7f1c16b400059, []int{7} +} +func (m *Err) XXX_Unmarshal(b []byte) error { + return m.Unmarshal(b) +} +func (m *Err) XXX_Marshal(b []byte, deterministic bool) ([]byte, error) { + if deterministic { + return xxx_messageInfo_Err.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 *Err) XXX_Merge(src proto.Message) { + xxx_messageInfo_Err.Merge(m, src) +} +func (m *Err) XXX_Size() int { + return m.Size() +} +func (m *Err) XXX_DiscardUnknown() { + xxx_messageInfo_Err.DiscardUnknown(m) +} + +var xxx_messageInfo_Err proto.InternalMessageInfo + +func (m *Err) GetError() ErrCodes { + if m != nil { + return m.Error + } + return ErrCodes_Unexpected +} + +func init() { + proto.RegisterEnum("consensusProto.ErrCodes", ErrCodes_name, ErrCodes_value) + proto.RegisterType((*Log)(nil), "consensusProto.Log") + proto.RegisterType((*Record)(nil), "consensusProto.Record") + proto.RegisterType((*Ok)(nil), "consensusProto.Ok") + proto.RegisterType((*LogAddRequest)(nil), "consensusProto.LogAddRequest") + proto.RegisterType((*RecordAddRequest)(nil), "consensusProto.RecordAddRequest") + proto.RegisterType((*LogWatchRequest)(nil), "consensusProto.LogWatchRequest") + proto.RegisterType((*LogWatchEvent)(nil), "consensusProto.LogWatchEvent") + proto.RegisterType((*Err)(nil), "consensusProto.Err") +} + +func init() { + proto.RegisterFile("consensus/consensusproto/protos/consensus.proto", fileDescriptor_b8d7f1c16b400059) +} + +var fileDescriptor_b8d7f1c16b400059 = []byte{ + // 511 bytes of a gzipped FileDescriptorProto + 0x1f, 0x8b, 0x08, 0x00, 0x00, 0x00, 0x00, 0x00, 0x02, 0xff, 0x94, 0x53, 0xc1, 0x6e, 0xd3, 0x40, + 0x10, 0xcd, 0xda, 0x6d, 0x9a, 0x4c, 0xda, 0xd4, 0x1a, 0x50, 0x65, 0x45, 0xaa, 0xb1, 0x2c, 0x21, + 0x05, 0x0e, 0x69, 0x65, 0x04, 0x17, 0x4e, 0x25, 0x32, 0x28, 0x92, 0x69, 0x90, 0xa5, 0x0a, 0xc4, + 0x89, 0xe0, 0xdd, 0x18, 0xab, 0x96, 0x37, 0xec, 0x3a, 0x25, 0x5c, 0xf9, 0x02, 0x3e, 0x84, 0x0f, + 0xe1, 0xd8, 0x23, 0x12, 0x17, 0x94, 0xfc, 0x08, 0xf2, 0x3a, 0x4e, 0x4d, 0x71, 0x84, 0xb8, 0xd8, + 0x7e, 0xef, 0xcd, 0xcc, 0xbe, 0x9d, 0x19, 0xc3, 0x49, 0xc8, 0x53, 0xc9, 0x52, 0x39, 0x97, 0x37, + 0x5f, 0x33, 0xc1, 0x33, 0x7e, 0xa2, 0x9e, 0x15, 0x76, 0xa0, 0x08, 0xec, 0x6e, 0x88, 0x57, 0x39, + 0x76, 0x5e, 0x80, 0xee, 0xf3, 0x08, 0xbb, 0xa0, 0xc5, 0xd4, 0x24, 0x36, 0xe9, 0xef, 0x07, 0x5a, + 0x4c, 0xf1, 0x14, 0xf6, 0x04, 0x0b, 0xb9, 0xa0, 0xd2, 0xd4, 0x6c, 0xbd, 0xdf, 0x71, 0x8f, 0x06, + 0x7f, 0x26, 0x0e, 0x02, 0x25, 0x07, 0x65, 0x98, 0x93, 0x40, 0xb3, 0xa0, 0xfe, 0xaa, 0x75, 0x04, + 0xcd, 0x99, 0x60, 0x57, 0x23, 0x6a, 0x6a, 0x8a, 0x5b, 0x23, 0x34, 0x61, 0x6f, 0x36, 0xf9, 0x9c, + 0xf0, 0x09, 0x35, 0x75, 0x25, 0x94, 0x10, 0x6d, 0xe8, 0x84, 0x82, 0x4d, 0x32, 0x46, 0x2f, 0xd2, + 0x78, 0x61, 0xee, 0xd8, 0xa4, 0xbf, 0x13, 0x54, 0x29, 0x67, 0x07, 0xb4, 0xf1, 0xa5, 0xf3, 0x04, + 0x0e, 0x7c, 0x1e, 0x9d, 0x51, 0x1a, 0xb0, 0x8f, 0x73, 0x26, 0x33, 0xbc, 0x0f, 0x7a, 0xc2, 0x23, + 0x75, 0x76, 0xc7, 0xbd, 0x73, 0xdb, 0xb2, 0xcf, 0xa3, 0x20, 0xd7, 0x9d, 0x37, 0x60, 0x14, 0x5e, + 0x2b, 0xa9, 0x77, 0x61, 0x37, 0xe1, 0xd1, 0xa8, 0x34, 0x5e, 0x00, 0x1c, 0x40, 0xb3, 0xb8, 0xa0, + 0xf2, 0xbe, 0xbd, 0x0d, 0xeb, 0x28, 0xe7, 0x25, 0x1c, 0xfa, 0x3c, 0x7a, 0x3d, 0xc9, 0xc2, 0x0f, + 0x65, 0xe1, 0x1e, 0xb4, 0x3e, 0xe5, 0x78, 0x44, 0xa5, 0x49, 0x6c, 0xbd, 0xbf, 0x1f, 0x6c, 0x30, + 0x5a, 0x00, 0xf3, 0x74, 0xa3, 0x6a, 0x4a, 0xad, 0x30, 0xce, 0x17, 0xa2, 0x6e, 0xa8, 0xea, 0x79, + 0x57, 0x2c, 0xdd, 0x66, 0xf3, 0xbf, 0xc7, 0x85, 0x0f, 0x60, 0x97, 0x09, 0xc1, 0x85, 0x6a, 0x7d, + 0x4d, 0xaf, 0x3c, 0x21, 0x82, 0x22, 0xc2, 0x79, 0x0c, 0xba, 0x27, 0x04, 0x0e, 0xca, 0x8c, 0xfc, + 0xe4, 0xae, 0x6b, 0xd6, 0x64, 0x0c, 0x39, 0x65, 0x72, 0x9d, 0xf6, 0xf0, 0x1d, 0xb4, 0x4a, 0x0a, + 0xbb, 0x00, 0x17, 0x29, 0x5b, 0xcc, 0x58, 0x98, 0x31, 0x6a, 0x34, 0xf0, 0x00, 0xda, 0x3e, 0x8f, + 0xbc, 0x45, 0x2c, 0x33, 0x69, 0x10, 0x3c, 0x84, 0x8e, 0xcf, 0xa3, 0x73, 0x9e, 0x3d, 0xe7, 0xf3, + 0x94, 0x1a, 0x1a, 0x22, 0x74, 0x0b, 0xc3, 0x43, 0x9e, 0x4e, 0x93, 0x38, 0xcc, 0x0c, 0x1d, 0x0d, + 0xe8, 0x78, 0x79, 0xe1, 0xf1, 0x74, 0x2a, 0x59, 0x66, 0x7c, 0xd3, 0xdc, 0x9f, 0x04, 0xda, 0xc3, + 0xd2, 0x04, 0x3e, 0x85, 0x66, 0xb1, 0x0c, 0x78, 0x5c, 0x33, 0xf8, 0x9b, 0x49, 0xf7, 0xf0, 0xb6, + 0x3c, 0xbe, 0xc4, 0x33, 0x68, 0x6f, 0x36, 0x02, 0xed, 0xfa, 0xe6, 0xfd, 0xa3, 0xc4, 0x39, 0xb4, + 0xca, 0x51, 0xe1, 0xbd, 0x1a, 0x07, 0xd5, 0xa5, 0xe8, 0x1d, 0x6f, 0x0b, 0x50, 0x53, 0xee, 0x93, + 0x53, 0xf2, 0xcc, 0xfd, 0xbe, 0xb4, 0xc8, 0xf5, 0xd2, 0x22, 0xbf, 0x96, 0x16, 0xf9, 0xba, 0xb2, + 0x1a, 0xd7, 0x2b, 0xab, 0xf1, 0x63, 0x65, 0x35, 0xde, 0x9a, 0xdb, 0x7e, 0xfa, 0xf7, 0x4d, 0xf5, + 0x7a, 0xf4, 0x3b, 0x00, 0x00, 0xff, 0xff, 0x8c, 0x81, 0x82, 0x9d, 0x17, 0x04, 0x00, 0x00, +} + +func (m *Log) 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 *Log) MarshalTo(dAtA []byte) (int, error) { + size := m.Size() + return m.MarshalToSizedBuffer(dAtA[:size]) +} + +func (m *Log) MarshalToSizedBuffer(dAtA []byte) (int, error) { + i := len(dAtA) + _ = i + var l int + _ = l + if len(m.Records) > 0 { + for iNdEx := len(m.Records) - 1; iNdEx >= 0; iNdEx-- { + { + size, err := m.Records[iNdEx].MarshalToSizedBuffer(dAtA[:i]) + if err != nil { + return 0, err + } + i -= size + i = encodeVarintConsensus(dAtA, i, uint64(size)) + } + i-- + dAtA[i] = 0x12 + } + } + if len(m.Id) > 0 { + i -= len(m.Id) + copy(dAtA[i:], m.Id) + i = encodeVarintConsensus(dAtA, i, uint64(len(m.Id))) + i-- + dAtA[i] = 0xa + } + return len(dAtA) - i, nil +} + +func (m *Record) 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 *Record) MarshalTo(dAtA []byte) (int, error) { + size := m.Size() + return m.MarshalToSizedBuffer(dAtA[:size]) +} + +func (m *Record) MarshalToSizedBuffer(dAtA []byte) (int, error) { + i := len(dAtA) + _ = i + var l int + _ = l + if m.CreatedUnix != 0 { + i = encodeVarintConsensus(dAtA, i, uint64(m.CreatedUnix)) + i-- + dAtA[i] = 0x20 + } + if len(m.Payload) > 0 { + i -= len(m.Payload) + copy(dAtA[i:], m.Payload) + i = encodeVarintConsensus(dAtA, i, uint64(len(m.Payload))) + i-- + dAtA[i] = 0x1a + } + if len(m.PrevId) > 0 { + i -= len(m.PrevId) + copy(dAtA[i:], m.PrevId) + i = encodeVarintConsensus(dAtA, i, uint64(len(m.PrevId))) + i-- + dAtA[i] = 0x12 + } + if len(m.Id) > 0 { + i -= len(m.Id) + copy(dAtA[i:], m.Id) + i = encodeVarintConsensus(dAtA, i, uint64(len(m.Id))) + i-- + dAtA[i] = 0xa + } + return len(dAtA) - i, nil +} + +func (m *Ok) 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 *Ok) MarshalTo(dAtA []byte) (int, error) { + size := m.Size() + return m.MarshalToSizedBuffer(dAtA[:size]) +} + +func (m *Ok) MarshalToSizedBuffer(dAtA []byte) (int, error) { + i := len(dAtA) + _ = i + var l int + _ = l + return len(dAtA) - i, nil +} + +func (m *LogAddRequest) 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 *LogAddRequest) MarshalTo(dAtA []byte) (int, error) { + size := m.Size() + return m.MarshalToSizedBuffer(dAtA[:size]) +} + +func (m *LogAddRequest) MarshalToSizedBuffer(dAtA []byte) (int, error) { + i := len(dAtA) + _ = i + var l int + _ = l + if m.Log != nil { + { + size, err := m.Log.MarshalToSizedBuffer(dAtA[:i]) + if err != nil { + return 0, err + } + i -= size + i = encodeVarintConsensus(dAtA, i, uint64(size)) + } + i-- + dAtA[i] = 0xa + } + return len(dAtA) - i, nil +} + +func (m *RecordAddRequest) 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 *RecordAddRequest) MarshalTo(dAtA []byte) (int, error) { + size := m.Size() + return m.MarshalToSizedBuffer(dAtA[:size]) +} + +func (m *RecordAddRequest) MarshalToSizedBuffer(dAtA []byte) (int, error) { + i := len(dAtA) + _ = i + var l int + _ = l + if m.Record != nil { + { + size, err := m.Record.MarshalToSizedBuffer(dAtA[:i]) + if err != nil { + return 0, err + } + i -= size + i = encodeVarintConsensus(dAtA, i, uint64(size)) + } + i-- + dAtA[i] = 0x12 + } + if len(m.LogId) > 0 { + i -= len(m.LogId) + copy(dAtA[i:], m.LogId) + i = encodeVarintConsensus(dAtA, i, uint64(len(m.LogId))) + i-- + dAtA[i] = 0xa + } + return len(dAtA) - i, nil +} + +func (m *LogWatchRequest) 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 *LogWatchRequest) MarshalTo(dAtA []byte) (int, error) { + size := m.Size() + return m.MarshalToSizedBuffer(dAtA[:size]) +} + +func (m *LogWatchRequest) MarshalToSizedBuffer(dAtA []byte) (int, error) { + i := len(dAtA) + _ = i + var l int + _ = l + if len(m.UnwatchIds) > 0 { + for iNdEx := len(m.UnwatchIds) - 1; iNdEx >= 0; iNdEx-- { + i -= len(m.UnwatchIds[iNdEx]) + copy(dAtA[i:], m.UnwatchIds[iNdEx]) + i = encodeVarintConsensus(dAtA, i, uint64(len(m.UnwatchIds[iNdEx]))) + i-- + dAtA[i] = 0x12 + } + } + if len(m.WatchIds) > 0 { + for iNdEx := len(m.WatchIds) - 1; iNdEx >= 0; iNdEx-- { + i -= len(m.WatchIds[iNdEx]) + copy(dAtA[i:], m.WatchIds[iNdEx]) + i = encodeVarintConsensus(dAtA, i, uint64(len(m.WatchIds[iNdEx]))) + i-- + dAtA[i] = 0xa + } + } + return len(dAtA) - i, nil +} + +func (m *LogWatchEvent) 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 *LogWatchEvent) MarshalTo(dAtA []byte) (int, error) { + size := m.Size() + return m.MarshalToSizedBuffer(dAtA[:size]) +} + +func (m *LogWatchEvent) MarshalToSizedBuffer(dAtA []byte) (int, error) { + i := len(dAtA) + _ = i + var l int + _ = l + if m.Error != nil { + { + size, err := m.Error.MarshalToSizedBuffer(dAtA[:i]) + if err != nil { + return 0, err + } + i -= size + i = encodeVarintConsensus(dAtA, i, uint64(size)) + } + i-- + dAtA[i] = 0x1a + } + if len(m.Records) > 0 { + for iNdEx := len(m.Records) - 1; iNdEx >= 0; iNdEx-- { + { + size, err := m.Records[iNdEx].MarshalToSizedBuffer(dAtA[:i]) + if err != nil { + return 0, err + } + i -= size + i = encodeVarintConsensus(dAtA, i, uint64(size)) + } + i-- + dAtA[i] = 0x12 + } + } + if len(m.LogId) > 0 { + i -= len(m.LogId) + copy(dAtA[i:], m.LogId) + i = encodeVarintConsensus(dAtA, i, uint64(len(m.LogId))) + i-- + dAtA[i] = 0xa + } + return len(dAtA) - i, nil +} + +func (m *Err) 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 *Err) MarshalTo(dAtA []byte) (int, error) { + size := m.Size() + return m.MarshalToSizedBuffer(dAtA[:size]) +} + +func (m *Err) MarshalToSizedBuffer(dAtA []byte) (int, error) { + i := len(dAtA) + _ = i + var l int + _ = l + if m.Error != 0 { + i = encodeVarintConsensus(dAtA, i, uint64(m.Error)) + i-- + dAtA[i] = 0x8 + } + return len(dAtA) - i, nil +} + +func encodeVarintConsensus(dAtA []byte, offset int, v uint64) int { + offset -= sovConsensus(v) + base := offset + for v >= 1<<7 { + dAtA[offset] = uint8(v&0x7f | 0x80) + v >>= 7 + offset++ + } + dAtA[offset] = uint8(v) + return base +} +func (m *Log) Size() (n int) { + if m == nil { + return 0 + } + var l int + _ = l + l = len(m.Id) + if l > 0 { + n += 1 + l + sovConsensus(uint64(l)) + } + if len(m.Records) > 0 { + for _, e := range m.Records { + l = e.Size() + n += 1 + l + sovConsensus(uint64(l)) + } + } + return n +} + +func (m *Record) Size() (n int) { + if m == nil { + return 0 + } + var l int + _ = l + l = len(m.Id) + if l > 0 { + n += 1 + l + sovConsensus(uint64(l)) + } + l = len(m.PrevId) + if l > 0 { + n += 1 + l + sovConsensus(uint64(l)) + } + l = len(m.Payload) + if l > 0 { + n += 1 + l + sovConsensus(uint64(l)) + } + if m.CreatedUnix != 0 { + n += 1 + sovConsensus(uint64(m.CreatedUnix)) + } + return n +} + +func (m *Ok) Size() (n int) { + if m == nil { + return 0 + } + var l int + _ = l + return n +} + +func (m *LogAddRequest) Size() (n int) { + if m == nil { + return 0 + } + var l int + _ = l + if m.Log != nil { + l = m.Log.Size() + n += 1 + l + sovConsensus(uint64(l)) + } + return n +} + +func (m *RecordAddRequest) Size() (n int) { + if m == nil { + return 0 + } + var l int + _ = l + l = len(m.LogId) + if l > 0 { + n += 1 + l + sovConsensus(uint64(l)) + } + if m.Record != nil { + l = m.Record.Size() + n += 1 + l + sovConsensus(uint64(l)) + } + return n +} + +func (m *LogWatchRequest) Size() (n int) { + if m == nil { + return 0 + } + var l int + _ = l + if len(m.WatchIds) > 0 { + for _, b := range m.WatchIds { + l = len(b) + n += 1 + l + sovConsensus(uint64(l)) + } + } + if len(m.UnwatchIds) > 0 { + for _, b := range m.UnwatchIds { + l = len(b) + n += 1 + l + sovConsensus(uint64(l)) + } + } + return n +} + +func (m *LogWatchEvent) Size() (n int) { + if m == nil { + return 0 + } + var l int + _ = l + l = len(m.LogId) + if l > 0 { + n += 1 + l + sovConsensus(uint64(l)) + } + if len(m.Records) > 0 { + for _, e := range m.Records { + l = e.Size() + n += 1 + l + sovConsensus(uint64(l)) + } + } + if m.Error != nil { + l = m.Error.Size() + n += 1 + l + sovConsensus(uint64(l)) + } + return n +} + +func (m *Err) Size() (n int) { + if m == nil { + return 0 + } + var l int + _ = l + if m.Error != 0 { + n += 1 + sovConsensus(uint64(m.Error)) + } + return n +} + +func sovConsensus(x uint64) (n int) { + return (math_bits.Len64(x|1) + 6) / 7 +} +func sozConsensus(x uint64) (n int) { + return sovConsensus(uint64((x << 1) ^ uint64((int64(x) >> 63)))) +} +func (m *Log) 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 ErrIntOverflowConsensus + } + 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: Log: wiretype end group for non-group") + } + if fieldNum <= 0 { + return fmt.Errorf("proto: Log: 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 byteLen int + for shift := uint(0); ; shift += 7 { + if shift >= 64 { + return ErrIntOverflowConsensus + } + if iNdEx >= l { + return io.ErrUnexpectedEOF + } + b := dAtA[iNdEx] + iNdEx++ + byteLen |= int(b&0x7F) << shift + if b < 0x80 { + break + } + } + if byteLen < 0 { + return ErrInvalidLengthConsensus + } + postIndex := iNdEx + byteLen + if postIndex < 0 { + return ErrInvalidLengthConsensus + } + if postIndex > l { + return io.ErrUnexpectedEOF + } + m.Id = append(m.Id[:0], dAtA[iNdEx:postIndex]...) + if m.Id == nil { + m.Id = []byte{} + } + iNdEx = postIndex + case 2: + if wireType != 2 { + return fmt.Errorf("proto: wrong wireType = %d for field Records", wireType) + } + var msglen int + for shift := uint(0); ; shift += 7 { + if shift >= 64 { + return ErrIntOverflowConsensus + } + if iNdEx >= l { + return io.ErrUnexpectedEOF + } + b := dAtA[iNdEx] + iNdEx++ + msglen |= int(b&0x7F) << shift + if b < 0x80 { + break + } + } + if msglen < 0 { + return ErrInvalidLengthConsensus + } + postIndex := iNdEx + msglen + if postIndex < 0 { + return ErrInvalidLengthConsensus + } + if postIndex > l { + return io.ErrUnexpectedEOF + } + m.Records = append(m.Records, &Record{}) + if err := m.Records[len(m.Records)-1].Unmarshal(dAtA[iNdEx:postIndex]); err != nil { + return err + } + iNdEx = postIndex + default: + iNdEx = preIndex + skippy, err := skipConsensus(dAtA[iNdEx:]) + if err != nil { + return err + } + if (skippy < 0) || (iNdEx+skippy) < 0 { + return ErrInvalidLengthConsensus + } + if (iNdEx + skippy) > l { + return io.ErrUnexpectedEOF + } + iNdEx += skippy + } + } + + if iNdEx > l { + return io.ErrUnexpectedEOF + } + return nil +} +func (m *Record) 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 ErrIntOverflowConsensus + } + 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: Record: wiretype end group for non-group") + } + if fieldNum <= 0 { + return fmt.Errorf("proto: Record: 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 byteLen int + for shift := uint(0); ; shift += 7 { + if shift >= 64 { + return ErrIntOverflowConsensus + } + if iNdEx >= l { + return io.ErrUnexpectedEOF + } + b := dAtA[iNdEx] + iNdEx++ + byteLen |= int(b&0x7F) << shift + if b < 0x80 { + break + } + } + if byteLen < 0 { + return ErrInvalidLengthConsensus + } + postIndex := iNdEx + byteLen + if postIndex < 0 { + return ErrInvalidLengthConsensus + } + if postIndex > l { + return io.ErrUnexpectedEOF + } + m.Id = append(m.Id[:0], dAtA[iNdEx:postIndex]...) + if m.Id == nil { + m.Id = []byte{} + } + iNdEx = postIndex + case 2: + if wireType != 2 { + return fmt.Errorf("proto: wrong wireType = %d for field PrevId", wireType) + } + var byteLen int + for shift := uint(0); ; shift += 7 { + if shift >= 64 { + return ErrIntOverflowConsensus + } + if iNdEx >= l { + return io.ErrUnexpectedEOF + } + b := dAtA[iNdEx] + iNdEx++ + byteLen |= int(b&0x7F) << shift + if b < 0x80 { + break + } + } + if byteLen < 0 { + return ErrInvalidLengthConsensus + } + postIndex := iNdEx + byteLen + if postIndex < 0 { + return ErrInvalidLengthConsensus + } + if postIndex > l { + return io.ErrUnexpectedEOF + } + m.PrevId = append(m.PrevId[:0], dAtA[iNdEx:postIndex]...) + if m.PrevId == nil { + m.PrevId = []byte{} + } + iNdEx = postIndex + case 3: + if wireType != 2 { + return fmt.Errorf("proto: wrong wireType = %d for field Payload", wireType) + } + var byteLen int + for shift := uint(0); ; shift += 7 { + if shift >= 64 { + return ErrIntOverflowConsensus + } + if iNdEx >= l { + return io.ErrUnexpectedEOF + } + b := dAtA[iNdEx] + iNdEx++ + byteLen |= int(b&0x7F) << shift + if b < 0x80 { + break + } + } + if byteLen < 0 { + return ErrInvalidLengthConsensus + } + postIndex := iNdEx + byteLen + if postIndex < 0 { + return ErrInvalidLengthConsensus + } + if postIndex > l { + return io.ErrUnexpectedEOF + } + m.Payload = append(m.Payload[:0], dAtA[iNdEx:postIndex]...) + if m.Payload == nil { + m.Payload = []byte{} + } + iNdEx = postIndex + case 4: + if wireType != 0 { + return fmt.Errorf("proto: wrong wireType = %d for field CreatedUnix", wireType) + } + m.CreatedUnix = 0 + for shift := uint(0); ; shift += 7 { + if shift >= 64 { + return ErrIntOverflowConsensus + } + if iNdEx >= l { + return io.ErrUnexpectedEOF + } + b := dAtA[iNdEx] + iNdEx++ + m.CreatedUnix |= uint64(b&0x7F) << shift + if b < 0x80 { + break + } + } + default: + iNdEx = preIndex + skippy, err := skipConsensus(dAtA[iNdEx:]) + if err != nil { + return err + } + if (skippy < 0) || (iNdEx+skippy) < 0 { + return ErrInvalidLengthConsensus + } + if (iNdEx + skippy) > l { + return io.ErrUnexpectedEOF + } + iNdEx += skippy + } + } + + if iNdEx > l { + return io.ErrUnexpectedEOF + } + return nil +} +func (m *Ok) 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 ErrIntOverflowConsensus + } + 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: Ok: wiretype end group for non-group") + } + if fieldNum <= 0 { + return fmt.Errorf("proto: Ok: illegal tag %d (wire type %d)", fieldNum, wire) + } + switch fieldNum { + default: + iNdEx = preIndex + skippy, err := skipConsensus(dAtA[iNdEx:]) + if err != nil { + return err + } + if (skippy < 0) || (iNdEx+skippy) < 0 { + return ErrInvalidLengthConsensus + } + if (iNdEx + skippy) > l { + return io.ErrUnexpectedEOF + } + iNdEx += skippy + } + } + + if iNdEx > l { + return io.ErrUnexpectedEOF + } + return nil +} +func (m *LogAddRequest) 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 ErrIntOverflowConsensus + } + 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: LogAddRequest: wiretype end group for non-group") + } + if fieldNum <= 0 { + return fmt.Errorf("proto: LogAddRequest: illegal tag %d (wire type %d)", fieldNum, wire) + } + switch fieldNum { + case 1: + if wireType != 2 { + return fmt.Errorf("proto: wrong wireType = %d for field Log", wireType) + } + var msglen int + for shift := uint(0); ; shift += 7 { + if shift >= 64 { + return ErrIntOverflowConsensus + } + if iNdEx >= l { + return io.ErrUnexpectedEOF + } + b := dAtA[iNdEx] + iNdEx++ + msglen |= int(b&0x7F) << shift + if b < 0x80 { + break + } + } + if msglen < 0 { + return ErrInvalidLengthConsensus + } + postIndex := iNdEx + msglen + if postIndex < 0 { + return ErrInvalidLengthConsensus + } + if postIndex > l { + return io.ErrUnexpectedEOF + } + if m.Log == nil { + m.Log = &Log{} + } + if err := m.Log.Unmarshal(dAtA[iNdEx:postIndex]); err != nil { + return err + } + iNdEx = postIndex + default: + iNdEx = preIndex + skippy, err := skipConsensus(dAtA[iNdEx:]) + if err != nil { + return err + } + if (skippy < 0) || (iNdEx+skippy) < 0 { + return ErrInvalidLengthConsensus + } + if (iNdEx + skippy) > l { + return io.ErrUnexpectedEOF + } + iNdEx += skippy + } + } + + if iNdEx > l { + return io.ErrUnexpectedEOF + } + return nil +} +func (m *RecordAddRequest) 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 ErrIntOverflowConsensus + } + 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: RecordAddRequest: wiretype end group for non-group") + } + if fieldNum <= 0 { + return fmt.Errorf("proto: RecordAddRequest: illegal tag %d (wire type %d)", fieldNum, wire) + } + switch fieldNum { + case 1: + if wireType != 2 { + return fmt.Errorf("proto: wrong wireType = %d for field LogId", wireType) + } + var byteLen int + for shift := uint(0); ; shift += 7 { + if shift >= 64 { + return ErrIntOverflowConsensus + } + if iNdEx >= l { + return io.ErrUnexpectedEOF + } + b := dAtA[iNdEx] + iNdEx++ + byteLen |= int(b&0x7F) << shift + if b < 0x80 { + break + } + } + if byteLen < 0 { + return ErrInvalidLengthConsensus + } + postIndex := iNdEx + byteLen + if postIndex < 0 { + return ErrInvalidLengthConsensus + } + if postIndex > l { + return io.ErrUnexpectedEOF + } + m.LogId = append(m.LogId[:0], dAtA[iNdEx:postIndex]...) + if m.LogId == nil { + m.LogId = []byte{} + } + iNdEx = postIndex + case 2: + if wireType != 2 { + return fmt.Errorf("proto: wrong wireType = %d for field Record", wireType) + } + var msglen int + for shift := uint(0); ; shift += 7 { + if shift >= 64 { + return ErrIntOverflowConsensus + } + if iNdEx >= l { + return io.ErrUnexpectedEOF + } + b := dAtA[iNdEx] + iNdEx++ + msglen |= int(b&0x7F) << shift + if b < 0x80 { + break + } + } + if msglen < 0 { + return ErrInvalidLengthConsensus + } + postIndex := iNdEx + msglen + if postIndex < 0 { + return ErrInvalidLengthConsensus + } + if postIndex > l { + return io.ErrUnexpectedEOF + } + if m.Record == nil { + m.Record = &Record{} + } + if err := m.Record.Unmarshal(dAtA[iNdEx:postIndex]); err != nil { + return err + } + iNdEx = postIndex + default: + iNdEx = preIndex + skippy, err := skipConsensus(dAtA[iNdEx:]) + if err != nil { + return err + } + if (skippy < 0) || (iNdEx+skippy) < 0 { + return ErrInvalidLengthConsensus + } + if (iNdEx + skippy) > l { + return io.ErrUnexpectedEOF + } + iNdEx += skippy + } + } + + if iNdEx > l { + return io.ErrUnexpectedEOF + } + return nil +} +func (m *LogWatchRequest) 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 ErrIntOverflowConsensus + } + 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: LogWatchRequest: wiretype end group for non-group") + } + if fieldNum <= 0 { + return fmt.Errorf("proto: LogWatchRequest: illegal tag %d (wire type %d)", fieldNum, wire) + } + switch fieldNum { + case 1: + if wireType != 2 { + return fmt.Errorf("proto: wrong wireType = %d for field WatchIds", wireType) + } + var byteLen int + for shift := uint(0); ; shift += 7 { + if shift >= 64 { + return ErrIntOverflowConsensus + } + if iNdEx >= l { + return io.ErrUnexpectedEOF + } + b := dAtA[iNdEx] + iNdEx++ + byteLen |= int(b&0x7F) << shift + if b < 0x80 { + break + } + } + if byteLen < 0 { + return ErrInvalidLengthConsensus + } + postIndex := iNdEx + byteLen + if postIndex < 0 { + return ErrInvalidLengthConsensus + } + if postIndex > l { + return io.ErrUnexpectedEOF + } + m.WatchIds = append(m.WatchIds, make([]byte, postIndex-iNdEx)) + copy(m.WatchIds[len(m.WatchIds)-1], dAtA[iNdEx:postIndex]) + iNdEx = postIndex + case 2: + if wireType != 2 { + return fmt.Errorf("proto: wrong wireType = %d for field UnwatchIds", wireType) + } + var byteLen int + for shift := uint(0); ; shift += 7 { + if shift >= 64 { + return ErrIntOverflowConsensus + } + if iNdEx >= l { + return io.ErrUnexpectedEOF + } + b := dAtA[iNdEx] + iNdEx++ + byteLen |= int(b&0x7F) << shift + if b < 0x80 { + break + } + } + if byteLen < 0 { + return ErrInvalidLengthConsensus + } + postIndex := iNdEx + byteLen + if postIndex < 0 { + return ErrInvalidLengthConsensus + } + if postIndex > l { + return io.ErrUnexpectedEOF + } + m.UnwatchIds = append(m.UnwatchIds, make([]byte, postIndex-iNdEx)) + copy(m.UnwatchIds[len(m.UnwatchIds)-1], dAtA[iNdEx:postIndex]) + iNdEx = postIndex + default: + iNdEx = preIndex + skippy, err := skipConsensus(dAtA[iNdEx:]) + if err != nil { + return err + } + if (skippy < 0) || (iNdEx+skippy) < 0 { + return ErrInvalidLengthConsensus + } + if (iNdEx + skippy) > l { + return io.ErrUnexpectedEOF + } + iNdEx += skippy + } + } + + if iNdEx > l { + return io.ErrUnexpectedEOF + } + return nil +} +func (m *LogWatchEvent) 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 ErrIntOverflowConsensus + } + 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: LogWatchEvent: wiretype end group for non-group") + } + if fieldNum <= 0 { + return fmt.Errorf("proto: LogWatchEvent: illegal tag %d (wire type %d)", fieldNum, wire) + } + switch fieldNum { + case 1: + if wireType != 2 { + return fmt.Errorf("proto: wrong wireType = %d for field LogId", wireType) + } + var byteLen int + for shift := uint(0); ; shift += 7 { + if shift >= 64 { + return ErrIntOverflowConsensus + } + if iNdEx >= l { + return io.ErrUnexpectedEOF + } + b := dAtA[iNdEx] + iNdEx++ + byteLen |= int(b&0x7F) << shift + if b < 0x80 { + break + } + } + if byteLen < 0 { + return ErrInvalidLengthConsensus + } + postIndex := iNdEx + byteLen + if postIndex < 0 { + return ErrInvalidLengthConsensus + } + if postIndex > l { + return io.ErrUnexpectedEOF + } + m.LogId = append(m.LogId[:0], dAtA[iNdEx:postIndex]...) + if m.LogId == nil { + m.LogId = []byte{} + } + iNdEx = postIndex + case 2: + if wireType != 2 { + return fmt.Errorf("proto: wrong wireType = %d for field Records", wireType) + } + var msglen int + for shift := uint(0); ; shift += 7 { + if shift >= 64 { + return ErrIntOverflowConsensus + } + if iNdEx >= l { + return io.ErrUnexpectedEOF + } + b := dAtA[iNdEx] + iNdEx++ + msglen |= int(b&0x7F) << shift + if b < 0x80 { + break + } + } + if msglen < 0 { + return ErrInvalidLengthConsensus + } + postIndex := iNdEx + msglen + if postIndex < 0 { + return ErrInvalidLengthConsensus + } + if postIndex > l { + return io.ErrUnexpectedEOF + } + m.Records = append(m.Records, &Record{}) + if err := m.Records[len(m.Records)-1].Unmarshal(dAtA[iNdEx:postIndex]); err != nil { + return err + } + iNdEx = postIndex + case 3: + if wireType != 2 { + return fmt.Errorf("proto: wrong wireType = %d for field Error", wireType) + } + var msglen int + for shift := uint(0); ; shift += 7 { + if shift >= 64 { + return ErrIntOverflowConsensus + } + if iNdEx >= l { + return io.ErrUnexpectedEOF + } + b := dAtA[iNdEx] + iNdEx++ + msglen |= int(b&0x7F) << shift + if b < 0x80 { + break + } + } + if msglen < 0 { + return ErrInvalidLengthConsensus + } + postIndex := iNdEx + msglen + if postIndex < 0 { + return ErrInvalidLengthConsensus + } + if postIndex > l { + return io.ErrUnexpectedEOF + } + if m.Error == nil { + m.Error = &Err{} + } + if err := m.Error.Unmarshal(dAtA[iNdEx:postIndex]); err != nil { + return err + } + iNdEx = postIndex + default: + iNdEx = preIndex + skippy, err := skipConsensus(dAtA[iNdEx:]) + if err != nil { + return err + } + if (skippy < 0) || (iNdEx+skippy) < 0 { + return ErrInvalidLengthConsensus + } + if (iNdEx + skippy) > l { + return io.ErrUnexpectedEOF + } + iNdEx += skippy + } + } + + if iNdEx > l { + return io.ErrUnexpectedEOF + } + return nil +} +func (m *Err) 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 ErrIntOverflowConsensus + } + 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: Err: wiretype end group for non-group") + } + if fieldNum <= 0 { + return fmt.Errorf("proto: Err: illegal tag %d (wire type %d)", fieldNum, wire) + } + switch fieldNum { + case 1: + if wireType != 0 { + return fmt.Errorf("proto: wrong wireType = %d for field Error", wireType) + } + m.Error = 0 + for shift := uint(0); ; shift += 7 { + if shift >= 64 { + return ErrIntOverflowConsensus + } + if iNdEx >= l { + return io.ErrUnexpectedEOF + } + b := dAtA[iNdEx] + iNdEx++ + m.Error |= ErrCodes(b&0x7F) << shift + if b < 0x80 { + break + } + } + default: + iNdEx = preIndex + skippy, err := skipConsensus(dAtA[iNdEx:]) + if err != nil { + return err + } + if (skippy < 0) || (iNdEx+skippy) < 0 { + return ErrInvalidLengthConsensus + } + if (iNdEx + skippy) > l { + return io.ErrUnexpectedEOF + } + iNdEx += skippy + } + } + + if iNdEx > l { + return io.ErrUnexpectedEOF + } + return nil +} +func skipConsensus(dAtA []byte) (n int, err error) { + l := len(dAtA) + iNdEx := 0 + depth := 0 + for iNdEx < l { + var wire uint64 + for shift := uint(0); ; shift += 7 { + if shift >= 64 { + return 0, ErrIntOverflowConsensus + } + if iNdEx >= l { + return 0, io.ErrUnexpectedEOF + } + b := dAtA[iNdEx] + iNdEx++ + wire |= (uint64(b) & 0x7F) << shift + if b < 0x80 { + break + } + } + wireType := int(wire & 0x7) + switch wireType { + case 0: + for shift := uint(0); ; shift += 7 { + if shift >= 64 { + return 0, ErrIntOverflowConsensus + } + if iNdEx >= l { + return 0, io.ErrUnexpectedEOF + } + iNdEx++ + if dAtA[iNdEx-1] < 0x80 { + break + } + } + case 1: + iNdEx += 8 + case 2: + var length int + for shift := uint(0); ; shift += 7 { + if shift >= 64 { + return 0, ErrIntOverflowConsensus + } + if iNdEx >= l { + return 0, io.ErrUnexpectedEOF + } + b := dAtA[iNdEx] + iNdEx++ + length |= (int(b) & 0x7F) << shift + if b < 0x80 { + break + } + } + if length < 0 { + return 0, ErrInvalidLengthConsensus + } + iNdEx += length + case 3: + depth++ + case 4: + if depth == 0 { + return 0, ErrUnexpectedEndOfGroupConsensus + } + depth-- + case 5: + iNdEx += 4 + default: + return 0, fmt.Errorf("proto: illegal wireType %d", wireType) + } + if iNdEx < 0 { + return 0, ErrInvalidLengthConsensus + } + if depth == 0 { + return iNdEx, nil + } + } + return 0, io.ErrUnexpectedEOF +} + +var ( + ErrInvalidLengthConsensus = fmt.Errorf("proto: negative length found during unmarshaling") + ErrIntOverflowConsensus = fmt.Errorf("proto: integer overflow") + ErrUnexpectedEndOfGroupConsensus = fmt.Errorf("proto: unexpected end of group") +) diff --git a/consensus/consensusproto/consensus_drpc.pb.go b/consensus/consensusproto/consensus_drpc.pb.go new file mode 100644 index 00000000..be8927c6 --- /dev/null +++ b/consensus/consensusproto/consensus_drpc.pb.go @@ -0,0 +1,232 @@ +// Code generated by protoc-gen-go-drpc. DO NOT EDIT. +// protoc-gen-go-drpc version: v0.0.33 +// source: consensus/consensusproto/protos/consensus.proto + +package consensusproto + +import ( + bytes "bytes" + context "context" + errors "errors" + jsonpb "github.com/gogo/protobuf/jsonpb" + proto "github.com/gogo/protobuf/proto" + drpc "storj.io/drpc" + drpcerr "storj.io/drpc/drpcerr" +) + +type drpcEncoding_File_consensus_consensusproto_protos_consensus_proto struct{} + +func (drpcEncoding_File_consensus_consensusproto_protos_consensus_proto) Marshal(msg drpc.Message) ([]byte, error) { + return proto.Marshal(msg.(proto.Message)) +} + +func (drpcEncoding_File_consensus_consensusproto_protos_consensus_proto) Unmarshal(buf []byte, msg drpc.Message) error { + return proto.Unmarshal(buf, msg.(proto.Message)) +} + +func (drpcEncoding_File_consensus_consensusproto_protos_consensus_proto) JSONMarshal(msg drpc.Message) ([]byte, error) { + var buf bytes.Buffer + err := new(jsonpb.Marshaler).Marshal(&buf, msg.(proto.Message)) + if err != nil { + return nil, err + } + return buf.Bytes(), nil +} + +func (drpcEncoding_File_consensus_consensusproto_protos_consensus_proto) JSONUnmarshal(buf []byte, msg drpc.Message) error { + return jsonpb.Unmarshal(bytes.NewReader(buf), msg.(proto.Message)) +} + +type DRPCConsensusClient interface { + DRPCConn() drpc.Conn + + LogAdd(ctx context.Context, in *LogAddRequest) (*Ok, error) + RecordAdd(ctx context.Context, in *RecordAddRequest) (*Ok, error) + LogWatch(ctx context.Context) (DRPCConsensus_LogWatchClient, error) +} + +type drpcConsensusClient struct { + cc drpc.Conn +} + +func NewDRPCConsensusClient(cc drpc.Conn) DRPCConsensusClient { + return &drpcConsensusClient{cc} +} + +func (c *drpcConsensusClient) DRPCConn() drpc.Conn { return c.cc } + +func (c *drpcConsensusClient) LogAdd(ctx context.Context, in *LogAddRequest) (*Ok, error) { + out := new(Ok) + err := c.cc.Invoke(ctx, "/consensusProto.Consensus/LogAdd", drpcEncoding_File_consensus_consensusproto_protos_consensus_proto{}, in, out) + if err != nil { + return nil, err + } + return out, nil +} + +func (c *drpcConsensusClient) RecordAdd(ctx context.Context, in *RecordAddRequest) (*Ok, error) { + out := new(Ok) + err := c.cc.Invoke(ctx, "/consensusProto.Consensus/RecordAdd", drpcEncoding_File_consensus_consensusproto_protos_consensus_proto{}, in, out) + if err != nil { + return nil, err + } + return out, nil +} + +func (c *drpcConsensusClient) LogWatch(ctx context.Context) (DRPCConsensus_LogWatchClient, error) { + stream, err := c.cc.NewStream(ctx, "/consensusProto.Consensus/LogWatch", drpcEncoding_File_consensus_consensusproto_protos_consensus_proto{}) + if err != nil { + return nil, err + } + x := &drpcConsensus_LogWatchClient{stream} + return x, nil +} + +type DRPCConsensus_LogWatchClient interface { + drpc.Stream + Send(*LogWatchRequest) error + Recv() (*LogWatchEvent, error) +} + +type drpcConsensus_LogWatchClient struct { + drpc.Stream +} + +func (x *drpcConsensus_LogWatchClient) GetStream() drpc.Stream { + return x.Stream +} + +func (x *drpcConsensus_LogWatchClient) Send(m *LogWatchRequest) error { + return x.MsgSend(m, drpcEncoding_File_consensus_consensusproto_protos_consensus_proto{}) +} + +func (x *drpcConsensus_LogWatchClient) Recv() (*LogWatchEvent, error) { + m := new(LogWatchEvent) + if err := x.MsgRecv(m, drpcEncoding_File_consensus_consensusproto_protos_consensus_proto{}); err != nil { + return nil, err + } + return m, nil +} + +func (x *drpcConsensus_LogWatchClient) RecvMsg(m *LogWatchEvent) error { + return x.MsgRecv(m, drpcEncoding_File_consensus_consensusproto_protos_consensus_proto{}) +} + +type DRPCConsensusServer interface { + LogAdd(context.Context, *LogAddRequest) (*Ok, error) + RecordAdd(context.Context, *RecordAddRequest) (*Ok, error) + LogWatch(DRPCConsensus_LogWatchStream) error +} + +type DRPCConsensusUnimplementedServer struct{} + +func (s *DRPCConsensusUnimplementedServer) LogAdd(context.Context, *LogAddRequest) (*Ok, error) { + return nil, drpcerr.WithCode(errors.New("Unimplemented"), drpcerr.Unimplemented) +} + +func (s *DRPCConsensusUnimplementedServer) RecordAdd(context.Context, *RecordAddRequest) (*Ok, error) { + return nil, drpcerr.WithCode(errors.New("Unimplemented"), drpcerr.Unimplemented) +} + +func (s *DRPCConsensusUnimplementedServer) LogWatch(DRPCConsensus_LogWatchStream) error { + return drpcerr.WithCode(errors.New("Unimplemented"), drpcerr.Unimplemented) +} + +type DRPCConsensusDescription struct{} + +func (DRPCConsensusDescription) NumMethods() int { return 3 } + +func (DRPCConsensusDescription) Method(n int) (string, drpc.Encoding, drpc.Receiver, interface{}, bool) { + switch n { + case 0: + return "/consensusProto.Consensus/LogAdd", drpcEncoding_File_consensus_consensusproto_protos_consensus_proto{}, + func(srv interface{}, ctx context.Context, in1, in2 interface{}) (drpc.Message, error) { + return srv.(DRPCConsensusServer). + LogAdd( + ctx, + in1.(*LogAddRequest), + ) + }, DRPCConsensusServer.LogAdd, true + case 1: + return "/consensusProto.Consensus/RecordAdd", drpcEncoding_File_consensus_consensusproto_protos_consensus_proto{}, + func(srv interface{}, ctx context.Context, in1, in2 interface{}) (drpc.Message, error) { + return srv.(DRPCConsensusServer). + RecordAdd( + ctx, + in1.(*RecordAddRequest), + ) + }, DRPCConsensusServer.RecordAdd, true + case 2: + return "/consensusProto.Consensus/LogWatch", drpcEncoding_File_consensus_consensusproto_protos_consensus_proto{}, + func(srv interface{}, ctx context.Context, in1, in2 interface{}) (drpc.Message, error) { + return nil, srv.(DRPCConsensusServer). + LogWatch( + &drpcConsensus_LogWatchStream{in1.(drpc.Stream)}, + ) + }, DRPCConsensusServer.LogWatch, true + default: + return "", nil, nil, nil, false + } +} + +func DRPCRegisterConsensus(mux drpc.Mux, impl DRPCConsensusServer) error { + return mux.Register(impl, DRPCConsensusDescription{}) +} + +type DRPCConsensus_LogAddStream interface { + drpc.Stream + SendAndClose(*Ok) error +} + +type drpcConsensus_LogAddStream struct { + drpc.Stream +} + +func (x *drpcConsensus_LogAddStream) SendAndClose(m *Ok) error { + if err := x.MsgSend(m, drpcEncoding_File_consensus_consensusproto_protos_consensus_proto{}); err != nil { + return err + } + return x.CloseSend() +} + +type DRPCConsensus_RecordAddStream interface { + drpc.Stream + SendAndClose(*Ok) error +} + +type drpcConsensus_RecordAddStream struct { + drpc.Stream +} + +func (x *drpcConsensus_RecordAddStream) SendAndClose(m *Ok) error { + if err := x.MsgSend(m, drpcEncoding_File_consensus_consensusproto_protos_consensus_proto{}); err != nil { + return err + } + return x.CloseSend() +} + +type DRPCConsensus_LogWatchStream interface { + drpc.Stream + Send(*LogWatchEvent) error + Recv() (*LogWatchRequest, error) +} + +type drpcConsensus_LogWatchStream struct { + drpc.Stream +} + +func (x *drpcConsensus_LogWatchStream) Send(m *LogWatchEvent) error { + return x.MsgSend(m, drpcEncoding_File_consensus_consensusproto_protos_consensus_proto{}) +} + +func (x *drpcConsensus_LogWatchStream) Recv() (*LogWatchRequest, error) { + m := new(LogWatchRequest) + if err := x.MsgRecv(m, drpcEncoding_File_consensus_consensusproto_protos_consensus_proto{}); err != nil { + return nil, err + } + return m, nil +} + +func (x *drpcConsensus_LogWatchStream) RecvMsg(m *LogWatchRequest) error { + return x.MsgRecv(m, drpcEncoding_File_consensus_consensusproto_protos_consensus_proto{}) +} diff --git a/consensus/consensusproto/consensuserr/errors.go b/consensus/consensusproto/consensuserr/errors.go new file mode 100644 index 00000000..a7ba9f96 --- /dev/null +++ b/consensus/consensusproto/consensuserr/errors.go @@ -0,0 +1,16 @@ +package consensuserr + +import ( + "fmt" + "github.com/anyproto/any-sync/consensus/consensusproto" + "github.com/anyproto/any-sync/net/rpc/rpcerr" +) + +var ( + errGroup = rpcerr.ErrGroup(consensusproto.ErrCodes_ErrorOffset) + + ErrUnexpected = errGroup.Register(fmt.Errorf("unexpected consensus error"), uint64(consensusproto.ErrCodes_Unexpected)) + ErrConflict = errGroup.Register(fmt.Errorf("records conflict"), uint64(consensusproto.ErrCodes_RecordConflict)) + ErrLogExists = errGroup.Register(fmt.Errorf("log exists"), uint64(consensusproto.ErrCodes_LogExists)) + ErrLogNotFound = errGroup.Register(fmt.Errorf("log not found"), uint64(consensusproto.ErrCodes_LogNotFound)) +) diff --git a/consensus/consensusproto/protos/consensus.proto b/consensus/consensusproto/protos/consensus.proto new file mode 100644 index 00000000..d28cf1de --- /dev/null +++ b/consensus/consensusproto/protos/consensus.proto @@ -0,0 +1,60 @@ +syntax = "proto3"; +package consensusProto; + +option go_package = "consensus/consensusproto"; + +enum ErrCodes { + Unexpected = 0; + LogExists = 1; + LogNotFound = 2; + RecordConflict = 3; + ErrorOffset = 300; +} + + +message Log { + bytes id = 1; + repeated Record records = 2; +} + +message Record { + bytes id = 1; + bytes prevId = 2; + bytes payload = 3; + uint64 createdUnix = 4; +} + +service Consensus { + // AddLog adds new log to consensus + rpc LogAdd(LogAddRequest) returns (Ok); + // AddRecord adds new record to log + rpc RecordAdd(RecordAddRequest) returns (Ok); + // WatchLog fetches log and subscribes for a changes + rpc LogWatch(stream LogWatchRequest) returns (stream LogWatchEvent); +} + +message Ok {} + +message LogAddRequest { + Log log = 1; +} + +message RecordAddRequest { + bytes logId = 1; + Record record = 2; +} + +message LogWatchRequest { + repeated bytes watchIds = 1; + repeated bytes unwatchIds = 2; +} + +message LogWatchEvent { + bytes logId = 1; + repeated Record records = 2; + Err error = 3; +} + +message Err { + ErrCodes error = 1; +} \ No newline at end of file