benjamyn/any-sync
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge remote-tracking branch 'origin/consensus-client' into acl-sync-protocol

Browse Source 
# Conflicts:
#	consensus/consensusproto/consensus.pb.go
This commit is contained in:
mcrakhman 2023-07-03 17:16:43 +02:00
commit 8aa41da1ff
No known key found for this signature in database
GPG Key ID: DED12CFEF5B8396B
5 changed files with 131 additions and 138 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

32
consensus/consensusclient/client.go
View File

@ -38,11 +38,11 @@ type Service interface {
// AddLog adds new log to consensus servers // AddLog adds new log to consensus servers
AddLog(ctx context.Context, clog *consensusproto.Log) (err error) AddLog(ctx context.Context, clog *consensusproto.Log) (err error)
// AddRecord adds new record to consensus servers // AddRecord adds new record to consensus servers
AddRecord(ctx context.Context, logId []byte, clog *consensusproto.RawRecord) (record *consensusproto.RawRecordWithId, err error) AddRecord(ctx context.Context, logId string, clog *consensusproto.RawRecord) (record *consensusproto.RawRecordWithId, err error)
// Watch starts watching to given logId and calls watcher when any relative event received // Watch starts watching to given logId and calls watcher when any relative event received
Watch(logId []byte, w Watcher) (err error) Watch(logId string, w Watcher) (err error)
// UnWatch stops watching given logId and removes watcher // UnWatch stops watching given logId and removes watcher
UnWatch(logId []byte) (err error) UnWatch(logId string) (err error)
app.ComponentRunnable app.ComponentRunnable
} }
@ -97,7 +97,7 @@ func (s *service) AddLog(ctx context.Context, clog *consensusproto.Log) (err err
}) })
} }
func (s *service) AddRecord(ctx context.Context, logId []byte, clog *consensusproto.RawRecord) (record *consensusproto.RawRecordWithId, err error) { func (s *service) AddRecord(ctx context.Context, logId string, clog *consensusproto.RawRecord) (record *consensusproto.RawRecordWithId, err error) {
err = s.doClient(ctx, func(cl consensusproto.DRPCConsensusClient) error { err = s.doClient(ctx, func(cl consensusproto.DRPCConsensusClient) error {
if record, err = cl.RecordAdd(ctx, &consensusproto.RecordAddRequest{ if record, err = cl.RecordAdd(ctx, &consensusproto.RecordAddRequest{
LogId: logId, LogId: logId,
@ -110,30 +110,30 @@ func (s *service) AddRecord(ctx context.Context, logId []byte, clog *consensuspr
return return
} }
func (s *service) Watch(logId []byte, w Watcher) (err error) { func (s *service) Watch(logId string, w Watcher) (err error) {
s.mu.Lock() s.mu.Lock()
defer s.mu.Unlock() defer s.mu.Unlock()
if _, ok := s.watchers[string(logId)]; ok { if _, ok := s.watchers[logId]; ok {
return ErrWatcherExists return ErrWatcherExists
} }
s.watchers[string(logId)] = w s.watchers[logId] = w
if s.stream != nil { if s.stream != nil {
if wErr := s.stream.WatchIds([][]byte{logId}); wErr != nil { if wErr := s.stream.WatchIds([]string{logId}); wErr != nil {
log.Warn("WatchIds error", zap.Error(wErr)) log.Warn("WatchIds error", zap.Error(wErr))
} }
} }
return return
} }
func (s *service) UnWatch(logId []byte) (err error) { func (s *service) UnWatch(logId string) (err error) {
s.mu.Lock() s.mu.Lock()
defer s.mu.Unlock() defer s.mu.Unlock()
if _, ok := s.watchers[string(logId)]; !ok { if _, ok := s.watchers[logId]; !ok {
return ErrWatcherNotExists return ErrWatcherNotExists
} }
delete(s.watchers, string(logId)) delete(s.watchers, logId)
if s.stream != nil { if s.stream != nil {
if wErr := s.stream.UnwatchIds([][]byte{logId}); wErr != nil { if wErr := s.stream.UnwatchIds([]string{logId}); wErr != nil {
log.Warn("UnWatchIds error", zap.Error(wErr)) log.Warn("UnWatchIds error", zap.Error(wErr))
} }
} }
@ -182,9 +182,9 @@ func (s *service) streamWatcher() {
// collect ids and setup stream // collect ids and setup stream
s.mu.Lock() s.mu.Lock()
var logIds = make([][]byte, 0, len(s.watchers)) var logIds = make([]string, 0, len(s.watchers))
for id := range s.watchers { for id := range s.watchers {
logIds = append(logIds, []byte(id)) logIds = append(logIds, id)
} }
s.stream = st s.stream = st
s.mu.Unlock() s.mu.Unlock()
@ -213,14 +213,14 @@ func (s *service) streamReader() error {
return s.stream.Err() return s.stream.Err()
} }
for _, e := range events { for _, e := range events {
if w, ok := s.watchers[string(e.LogId)]; ok { if w, ok := s.watchers[e.LogId]; ok {
if e.Error == nil { if e.Error == nil {
w.AddConsensusRecords(e.Records) w.AddConsensusRecords(e.Records)
} else { } else {
w.AddConsensusError(rpcerr.Err(uint64(e.Error.Error))) w.AddConsensusError(rpcerr.Err(uint64(e.Error.Error)))
} }
} else { } else {
log.Warn("received unexpected log id", zap.Binary("logId", e.LogId)) log.Warn("received unexpected log id", zap.String("logId", e.LogId))
} }
} }
} }

22
consensus/consensusclient/client_test.go
View File

@ -24,13 +24,13 @@ func TestService_Watch(t *testing.T) {
t.Run("not found error", func(t *testing.T) { t.Run("not found error", func(t *testing.T) {
fx := newFixture(t).run(t) fx := newFixture(t).run(t)
defer fx.Finish() defer fx.Finish()
var logId = []byte{'1'} var logId = "1"
w := &testWatcher{ready: make(chan struct{})} w := &testWatcher{ready: make(chan struct{})}
require.NoError(t, fx.Watch(logId, w)) require.NoError(t, fx.Watch(logId, w))
st := fx.testServer.waitStream(t) st := fx.testServer.waitStream(t)
req, err := st.Recv() req, err := st.Recv()
require.NoError(t, err) require.NoError(t, err)
assert.Equal(t, [][]byte{logId}, req.WatchIds) assert.Equal(t, []string{logId}, req.WatchIds)
require.NoError(t, st.Send(&consensusproto.LogWatchEvent{ require.NoError(t, st.Send(&consensusproto.LogWatchEvent{
LogId: logId, LogId: logId,
Error: &consensusproto.Err{ Error: &consensusproto.Err{
@ -44,7 +44,7 @@ func TestService_Watch(t *testing.T) {
t.Run("watcherExists error", func(t *testing.T) { t.Run("watcherExists error", func(t *testing.T) {
fx := newFixture(t).run(t) fx := newFixture(t).run(t)
defer fx.Finish() defer fx.Finish()
var logId = []byte{'1'} var logId = "1"
w := &testWatcher{} w := &testWatcher{}
require.NoError(t, fx.Watch(logId, w)) require.NoError(t, fx.Watch(logId, w))
require.Error(t, fx.Watch(logId, w)) require.Error(t, fx.Watch(logId, w))
@ -55,20 +55,20 @@ func TestService_Watch(t *testing.T) {
t.Run("watch", func(t *testing.T) { t.Run("watch", func(t *testing.T) {
fx := newFixture(t).run(t) fx := newFixture(t).run(t)
defer fx.Finish() defer fx.Finish()
var logId1 = []byte{'1'} var logId1 = "1"
w := &testWatcher{} w := &testWatcher{}
require.NoError(t, fx.Watch(logId1, w)) require.NoError(t, fx.Watch(logId1, w))
st := fx.testServer.waitStream(t) st := fx.testServer.waitStream(t)
req, err := st.Recv() req, err := st.Recv()
require.NoError(t, err) require.NoError(t, err)
assert.Equal(t, [][]byte{logId1}, req.WatchIds) assert.Equal(t, []string{logId1}, req.WatchIds)
var logId2 = []byte{'2'} var logId2 = "2"
w = &testWatcher{} w = &testWatcher{}
require.NoError(t, fx.Watch(logId2, w)) require.NoError(t, fx.Watch(logId2, w))
req, err = st.Recv() req, err = st.Recv()
require.NoError(t, err) require.NoError(t, err)
assert.Equal(t, [][]byte{logId2}, req.WatchIds) assert.Equal(t, []string{logId2}, req.WatchIds)
fx.testServer.releaseStream <- nil fx.testServer.releaseStream <- nil
}) })
@ -78,14 +78,14 @@ func TestService_UnWatch(t *testing.T) {
t.Run("no watcher", func(t *testing.T) { t.Run("no watcher", func(t *testing.T) {
fx := newFixture(t).run(t) fx := newFixture(t).run(t)
defer fx.Finish() defer fx.Finish()
require.Error(t, fx.UnWatch([]byte{'1'})) require.Error(t, fx.UnWatch("1"))
}) })
t.Run("success", func(t *testing.T) { t.Run("success", func(t *testing.T) {
fx := newFixture(t).run(t) fx := newFixture(t).run(t)
defer fx.Finish() defer fx.Finish()
w := &testWatcher{} w := &testWatcher{}
require.NoError(t, fx.Watch([]byte{'1'}, w)) require.NoError(t, fx.Watch("1", w))
assert.NoError(t, fx.UnWatch([]byte{'1'})) assert.NoError(t, fx.UnWatch("1"))
}) })
} }
@ -119,7 +119,7 @@ func TestService_AddLog(t *testing.T) {
func TestService_AddRecord(t *testing.T) { func TestService_AddRecord(t *testing.T) {
fx := newFixture(t).run(t) fx := newFixture(t).run(t)
defer fx.Finish() defer fx.Finish()
rec, err := fx.AddRecord(ctx, []byte{'1'}, &consensusproto.RawRecord{}) rec, err := fx.AddRecord(ctx, "1", &consensusproto.RawRecord{})
require.NoError(t, err) require.NoError(t, err)
assert.NotEmpty(t, rec) assert.NotEmpty(t, rec)
} }

4
consensus/consensusclient/stream.go
View File

@ -23,13 +23,13 @@ type stream struct {
err error err error
} }
func (s *stream) WatchIds(logIds [][]byte) (err error) { func (s *stream) WatchIds(logIds []string) (err error) {
return s.rpcStream.Send(&consensusproto.LogWatchRequest{ return s.rpcStream.Send(&consensusproto.LogWatchRequest{
WatchIds: logIds, WatchIds: logIds,
}) })
} }
func (s *stream) UnwatchIds(logIds [][]byte) (err error) { func (s *stream) UnwatchIds(logIds []string) (err error) {
return s.rpcStream.Send(&consensusproto.LogWatchRequest{ return s.rpcStream.Send(&consensusproto.LogWatchRequest{
UnwatchIds: logIds, UnwatchIds: logIds,
}) })

201
consensus/consensusproto/consensus.pb.go
View File

@ -57,7 +57,7 @@ func (ErrCodes) EnumDescriptor() ([]byte, []int) {
} }
type Log struct { type Log struct {
Id []byte `protobuf:"bytes,1,opt,name=id,proto3" json:"id,omitempty"` Id string `protobuf:"bytes,1,opt,name=id,proto3" json:"id,omitempty"`
Payload []byte `protobuf:"bytes,2,opt,name=payload,proto3" json:"payload,omitempty"` Payload []byte `protobuf:"bytes,2,opt,name=payload,proto3" json:"payload,omitempty"`
Records []*RawRecordWithId `protobuf:"bytes,3,rep,name=records,proto3" json:"records,omitempty"` Records []*RawRecordWithId `protobuf:"bytes,3,rep,name=records,proto3" json:"records,omitempty"`
} }
@ -95,11 +95,11 @@ func (m *Log) XXX_DiscardUnknown() {
var xxx_messageInfo_Log proto.InternalMessageInfo var xxx_messageInfo_Log proto.InternalMessageInfo
func (m *Log) GetId() []byte { func (m *Log) GetId() string {
if m != nil { if m != nil {
return m.Id return m.Id
} }
return nil return ""
} }
func (m *Log) GetPayload() []byte { func (m *Log) GetPayload() []byte {
@ -388,7 +388,7 @@ func (m *LogAddRequest) GetLog() *Log {
} }
type RecordAddRequest struct { type RecordAddRequest struct {
LogId []byte `protobuf:"bytes,1,opt,name=logId,proto3" json:"logId,omitempty"` LogId string `protobuf:"bytes,1,opt,name=logId,proto3" json:"logId,omitempty"`
Record *RawRecord `protobuf:"bytes,2,opt,name=record,proto3" json:"record,omitempty"` Record *RawRecord `protobuf:"bytes,2,opt,name=record,proto3" json:"record,omitempty"`
} }
@ -425,11 +425,11 @@ func (m *RecordAddRequest) XXX_DiscardUnknown() {
var xxx_messageInfo_RecordAddRequest proto.InternalMessageInfo var xxx_messageInfo_RecordAddRequest proto.InternalMessageInfo
func (m *RecordAddRequest) GetLogId() []byte { func (m *RecordAddRequest) GetLogId() string {
if m != nil { if m != nil {
return m.LogId return m.LogId
} }
return nil return ""
} }
func (m *RecordAddRequest) GetRecord() *RawRecord { func (m *RecordAddRequest) GetRecord() *RawRecord {
@ -440,8 +440,8 @@ func (m *RecordAddRequest) GetRecord() *RawRecord {
} }
type LogWatchRequest struct { type LogWatchRequest struct {
WatchIds [][]byte `protobuf:"bytes,1,rep,name=watchIds,proto3" json:"watchIds,omitempty"` WatchIds []string `protobuf:"bytes,1,rep,name=watchIds,proto3" json:"watchIds,omitempty"`
UnwatchIds [][]byte `protobuf:"bytes,2,rep,name=unwatchIds,proto3" json:"unwatchIds,omitempty"` UnwatchIds []string `protobuf:"bytes,2,rep,name=unwatchIds,proto3" json:"unwatchIds,omitempty"`
} }
func (m *LogWatchRequest) Reset() { *m = LogWatchRequest{} } func (m *LogWatchRequest) Reset() { *m = LogWatchRequest{} }
@ -477,14 +477,14 @@ func (m *LogWatchRequest) XXX_DiscardUnknown() {
var xxx_messageInfo_LogWatchRequest proto.InternalMessageInfo var xxx_messageInfo_LogWatchRequest proto.InternalMessageInfo
func (m *LogWatchRequest) GetWatchIds() [][]byte { func (m *LogWatchRequest) GetWatchIds() []string {
if m != nil { if m != nil {
return m.WatchIds return m.WatchIds
} }
return nil return nil
} }
func (m *LogWatchRequest) GetUnwatchIds() [][]byte { func (m *LogWatchRequest) GetUnwatchIds() []string {
if m != nil { if m != nil {
return m.UnwatchIds return m.UnwatchIds
} }
@ -492,7 +492,7 @@ func (m *LogWatchRequest) GetUnwatchIds() [][]byte {
} }
type LogWatchEvent struct { type LogWatchEvent struct {
LogId []byte `protobuf:"bytes,1,opt,name=logId,proto3" json:"logId,omitempty"` LogId string `protobuf:"bytes,1,opt,name=logId,proto3" json:"logId,omitempty"`
Records []*RawRecordWithId `protobuf:"bytes,2,rep,name=records,proto3" json:"records,omitempty"` Records []*RawRecordWithId `protobuf:"bytes,2,rep,name=records,proto3" json:"records,omitempty"`
Error *Err `protobuf:"bytes,3,opt,name=error,proto3" json:"error,omitempty"` Error *Err `protobuf:"bytes,3,opt,name=error,proto3" json:"error,omitempty"`
} }
@ -530,11 +530,11 @@ func (m *LogWatchEvent) XXX_DiscardUnknown() {
var xxx_messageInfo_LogWatchEvent proto.InternalMessageInfo var xxx_messageInfo_LogWatchEvent proto.InternalMessageInfo
func (m *LogWatchEvent) GetLogId() []byte { func (m *LogWatchEvent) GetLogId() string {
if m != nil { if m != nil {
return m.LogId return m.LogId
} }
return nil return ""
} }
func (m *LogWatchEvent) GetRecords() []*RawRecordWithId { func (m *LogWatchEvent) GetRecords() []*RawRecordWithId {
@ -939,57 +939,56 @@ func init() {
} }
var fileDescriptor_b8d7f1c16b400059 = []byte{ var fileDescriptor_b8d7f1c16b400059 = []byte{
// 785 bytes of a gzipped FileDescriptorProto // 780 bytes of a gzipped FileDescriptorProto
0x1f, 0x8b, 0x08, 0x00, 0x00, 0x00, 0x00, 0x00, 0x02, 0xff, 0xac, 0x55, 0x5f, 0x4f, 0xdb, 0x56, 0x1f, 0x8b, 0x08, 0x00, 0x00, 0x00, 0x00, 0x00, 0x02, 0xff, 0xac, 0x55, 0xcd, 0x4e, 0xe3, 0x56,
0x14, 0x8f, 0x6d, 0x48, 0xe2, 0x93, 0x91, 0x78, 0x87, 0x69, 0xf2, 0xa2, 0x11, 0x22, 0x4f, 0x93, 0x14, 0x8e, 0x6d, 0x48, 0xe2, 0x93, 0x92, 0xb8, 0x87, 0xaa, 0x72, 0xa3, 0x12, 0x22, 0x57, 0x95,
0x18, 0x9a, 0xc2, 0x96, 0x69, 0x93, 0x26, 0x34, 0x4d, 0x23, 0x0a, 0x4a, 0xa4, 0x10, 0x36, 0x23, 0x52, 0x54, 0x85, 0x36, 0x55, 0x2b, 0x55, 0xa8, 0xaa, 0x4a, 0x14, 0x94, 0x48, 0x21, 0xb4, 0x46,
0x86, 0xb4, 0x49, 0x55, 0x5d, 0xdf, 0x1b, 0xe3, 0x12, 0x7c, 0x5d, 0xfb, 0x06, 0xc8, 0x6b, 0x5f, 0x14, 0xa9, 0x95, 0xaa, 0xba, 0xbe, 0x37, 0xc6, 0x25, 0xf8, 0xba, 0xf6, 0x0d, 0x90, 0x6d, 0xb7,
0xfb, 0xd2, 0x2f, 0xd0, 0xef, 0xd3, 0x47, 0x1e, 0xfb, 0x58, 0xc1, 0x4b, 0x3f, 0x40, 0x3f, 0x40, 0xb3, 0x99, 0x17, 0x98, 0xf7, 0x99, 0x25, 0xcb, 0x59, 0x8e, 0x60, 0x33, 0x0f, 0x30, 0x0f, 0x30,
0xe5, 0x6b, 0x3b, 0x31, 0xf9, 0x43, 0x55, 0x95, 0x17, 0xf0, 0xf9, 0xff, 0x3b, 0xbf, 0x73, 0xee, 0xf2, 0xf5, 0x4f, 0x42, 0x7e, 0x40, 0xa3, 0x61, 0x03, 0x3e, 0xff, 0xdf, 0xf9, 0xce, 0xb9, 0x27,
0x09, 0xec, 0xd8, 0xcc, 0x0b, 0xa9, 0x17, 0x8e, 0xc2, 0xe9, 0x97, 0x1f, 0x30, 0xce, 0x76, 0xc4, 0xb0, 0x6b, 0x33, 0x2f, 0xa4, 0x5e, 0x38, 0x0e, 0xa7, 0x5f, 0x7e, 0xc0, 0x38, 0xdb, 0x15, 0x7f,
0xdf, 0x8c, 0xb6, 0x21, 0x14, 0x58, 0x9e, 0x28, 0xfe, 0x8e, 0x64, 0xe3, 0x29, 0x28, 0x3d, 0xe6, 0x67, 0xb4, 0x4d, 0xa1, 0xc0, 0x72, 0xa6, 0xf8, 0x35, 0x92, 0x8d, 0x7f, 0x41, 0xe9, 0x33, 0x07,
0x60, 0x19, 0x64, 0x97, 0xe8, 0x52, 0x5d, 0xda, 0xfa, 0xc2, 0x94, 0x5d, 0x82, 0x3a, 0x14, 0x7c, 0xcb, 0x20, 0xbb, 0x44, 0x97, 0xea, 0x52, 0x43, 0x35, 0x65, 0x97, 0xa0, 0x0e, 0x05, 0xdf, 0x9a,
0x6b, 0x3c, 0x64, 0x16, 0xd1, 0x65, 0xa1, 0x4c, 0x45, 0xfc, 0x1d, 0x0a, 0x01, 0xb5, 0x59, 0x40, 0x8c, 0x98, 0x45, 0x74, 0xb9, 0x2e, 0x35, 0x3e, 0x32, 0x53, 0x11, 0x7f, 0x84, 0x42, 0x40, 0x6d,
0x42, 0x5d, 0xa9, 0x2b, 0x5b, 0xa5, 0xe6, 0x66, 0xe3, 0x6e, 0xca, 0x86, 0x69, 0x5d, 0x9a, 0xc2, 0x16, 0x90, 0x50, 0x57, 0xea, 0x4a, 0xa3, 0xd4, 0xda, 0x6e, 0xde, 0x4f, 0xd9, 0x34, 0xad, 0x2b,
0xe3, 0xc4, 0xe5, 0xa7, 0x5d, 0x62, 0xa6, 0xfe, 0xc6, 0x2b, 0x09, 0xd4, 0x89, 0x31, 0x5b, 0x42, 0x53, 0x78, 0x9c, 0xba, 0xfc, 0xac, 0x47, 0xcc, 0xd4, 0xdf, 0x78, 0x21, 0x81, 0x9a, 0x19, 0x67,
0xba, 0x5b, 0xe2, 0x5b, 0x50, 0x43, 0xd7, 0xf1, 0x2c, 0x3e, 0x0a, 0x68, 0x52, 0x7e, 0xaa, 0xc0, 0x4b, 0x48, 0xf7, 0x4b, 0x7c, 0x0e, 0x6a, 0xe8, 0x3a, 0x9e, 0xc5, 0xc7, 0x01, 0x4d, 0xca, 0x4f,
0x6d, 0xd0, 0x2c, 0xdb, 0xa6, 0x3e, 0x67, 0x41, 0x97, 0x50, 0x8f, 0xbb, 0x7c, 0xac, 0x2b, 0xc2, 0x15, 0xb8, 0x03, 0x9a, 0x65, 0xdb, 0xd4, 0xe7, 0x2c, 0xe8, 0x11, 0xea, 0x71, 0x97, 0x4f, 0x74,
0x69, 0x4e, 0x8f, 0x3f, 0xc2, 0x97, 0xa9, 0xee, 0x68, 0x92, 0x71, 0x45, 0x38, 0xcf, 0x1b, 0x8c, 0x45, 0x38, 0x2d, 0xe8, 0xf1, 0x6b, 0xf8, 0x38, 0xd5, 0x1d, 0x67, 0x19, 0xd7, 0x84, 0xf3, 0xa2,
0x5d, 0xa8, 0xcc, 0x60, 0xbf, 0x07, 0x64, 0xcc, 0x58, 0x84, 0x4e, 0x8d, 0x18, 0x33, 0x3c, 0xc8, 0xc1, 0xd8, 0x83, 0xca, 0x1c, 0xf6, 0x07, 0x40, 0xc6, 0x8c, 0xc9, 0x29, 0x63, 0x86, 0x07, 0xf9,
0x27, 0x8d, 0x7d, 0x0d, 0x79, 0x3f, 0xa0, 0x17, 0xdd, 0x38, 0x44, 0x35, 0x13, 0x09, 0xab, 0x50, 0xa4, 0xb1, 0x4f, 0x21, 0xef, 0x07, 0xf4, 0xb2, 0x97, 0xf2, 0x99, 0x48, 0x58, 0x85, 0xa2, 0x9b,
0x74, 0x53, 0xc0, 0x71, 0x57, 0x13, 0x19, 0x11, 0x56, 0x88, 0xc5, 0xad, 0xa4, 0x11, 0xf1, 0x1d, 0x02, 0x8e, 0xbb, 0xca, 0x64, 0x44, 0x58, 0x23, 0x16, 0xb7, 0x92, 0x46, 0xc4, 0x77, 0x44, 0x03,
0xd1, 0xc0, 0xdd, 0x73, 0x1a, 0x72, 0xeb, 0xdc, 0x17, 0xa0, 0x15, 0x73, 0xaa, 0x30, 0x56, 0x40, 0x77, 0x2f, 0x68, 0xc8, 0xad, 0x0b, 0x5f, 0x80, 0x56, 0xcc, 0xa9, 0xc2, 0x58, 0x03, 0xf9, 0xe8,
0x3e, 0x3c, 0x33, 0x7e, 0x83, 0xb5, 0x1e, 0x73, 0xfe, 0x22, 0xc4, 0xa4, 0xcf, 0x46, 0x34, 0xe4, 0xdc, 0xf8, 0x01, 0x36, 0xfa, 0xcc, 0xf9, 0x85, 0x10, 0x93, 0xfe, 0x37, 0xa6, 0x21, 0xc7, 0x2f,
0xf8, 0x3d, 0x28, 0x43, 0xe6, 0x88, 0xca, 0xa5, 0xe6, 0xfa, 0xec, 0x68, 0x7a, 0xcc, 0x31, 0x23, 0x41, 0x19, 0x31, 0x47, 0x54, 0x2e, 0xb5, 0x36, 0xe7, 0x47, 0xd3, 0x67, 0x8e, 0x19, 0xd9, 0x8d,
0xbb, 0xf1, 0x3f, 0x68, 0x31, 0xda, 0x4c, 0xe8, 0x57, 0xb0, 0x3a, 0x64, 0x4e, 0x37, 0xed, 0x34, 0x3f, 0x41, 0x8b, 0xd1, 0xce, 0x84, 0x7e, 0x02, 0xeb, 0x23, 0xe6, 0x64, 0xb0, 0x63, 0x01, 0xbf,
0x16, 0xf0, 0x67, 0xc8, 0xc7, 0xf3, 0x13, 0x98, 0x4b, 0xcd, 0x6f, 0x96, 0x8e, 0xdb, 0x4c, 0x1c, 0x85, 0x7c, 0x3c, 0x3f, 0x81, 0xb9, 0xd4, 0xfa, 0x6c, 0xe5, 0xb8, 0xcd, 0xc4, 0xd1, 0x38, 0x84,
0x8d, 0x03, 0xa8, 0xf4, 0x98, 0x73, 0x62, 0x71, 0xfb, 0x34, 0xcd, 0x5d, 0x85, 0xe2, 0x65, 0x24, 0x4a, 0x9f, 0x39, 0xa7, 0x16, 0xb7, 0xcf, 0xd2, 0xdc, 0x55, 0x28, 0x5e, 0x45, 0x72, 0x8f, 0x84,
0x77, 0x49, 0xa8, 0x4b, 0x75, 0x25, 0xea, 0x3d, 0x95, 0xb1, 0x06, 0x30, 0xf2, 0x26, 0x56, 0x59, 0xba, 0x54, 0x57, 0x1a, 0xaa, 0x99, 0xc9, 0x58, 0x03, 0x18, 0x7b, 0x99, 0x55, 0x16, 0xd6, 0x19,
0x58, 0x33, 0x1a, 0xe3, 0x85, 0x24, 0x9a, 0x14, 0xf9, 0xda, 0x17, 0xd4, 0x5b, 0x86, 0x34, 0xb3, 0x8d, 0xf1, 0x4c, 0x12, 0x4d, 0x8a, 0x7c, 0x9d, 0x4b, 0xea, 0xad, 0x42, 0x3a, 0xb3, 0x99, 0xf2,
0x99, 0xf2, 0xa7, 0x6d, 0x26, 0xfe, 0x00, 0xab, 0x34, 0x08, 0x58, 0x20, 0xf8, 0x5f, 0xc0, 0x5b, 0xfb, 0x6d, 0x26, 0x7e, 0x05, 0xeb, 0x34, 0x08, 0x58, 0x20, 0xf8, 0x5f, 0xc2, 0x5b, 0x27, 0x08,
0x3b, 0x08, 0xcc, 0xd8, 0xc3, 0xf8, 0x15, 0x94, 0x76, 0x10, 0x60, 0x23, 0x8d, 0x88, 0x20, 0x94, 0xcc, 0xd8, 0xc3, 0xf8, 0x1e, 0x94, 0x4e, 0x10, 0x60, 0x33, 0x8d, 0x88, 0x20, 0x94, 0x5b, 0xfa,
0x9b, 0xfa, 0x82, 0x88, 0x16, 0x23, 0x34, 0x4c, 0xc3, 0x9e, 0xcb, 0xb0, 0xde, 0x63, 0xce, 0xd1, 0x92, 0x88, 0x36, 0x23, 0x34, 0x4c, 0xc3, 0xfe, 0x97, 0x61, 0xb3, 0xcf, 0x9c, 0xe3, 0x89, 0x67,
0xd8, 0xb3, 0x5b, 0xcc, 0xe3, 0xd4, 0xe3, 0xff, 0x5a, 0xc3, 0x11, 0xc5, 0x3f, 0x01, 0x4e, 0xa9, 0xb7, 0x99, 0xc7, 0xa9, 0xc7, 0x7f, 0xb7, 0x46, 0x63, 0x8a, 0x3f, 0x03, 0x9c, 0x51, 0x8b, 0x9c,
0x45, 0x8e, 0x7d, 0x62, 0x71, 0x9a, 0x8c, 0x6d, 0x63, 0xc1, 0xd8, 0x3a, 0x13, 0xa7, 0x4e, 0xce, 0xf8, 0xc4, 0xe2, 0x34, 0x19, 0xdb, 0xd6, 0x92, 0xb1, 0x75, 0x33, 0xa7, 0x6e, 0xce, 0x9c, 0x09,
0xcc, 0x84, 0x60, 0x1f, 0x2a, 0x83, 0xd1, 0x70, 0x18, 0x25, 0x4e, 0xc8, 0x4e, 0x06, 0x65, 0x2c, 0xc1, 0x01, 0x54, 0x86, 0xe3, 0xd1, 0x28, 0x4a, 0x9c, 0x90, 0x9d, 0x0c, 0xca, 0x58, 0x92, 0xe5,
0xc8, 0xb2, 0x7f, 0xd7, 0xb3, 0x93, 0x33, 0x67, 0x83, 0xf1, 0x1f, 0xd0, 0xa6, 0xaa, 0xd0, 0x8f, 0xe0, 0xbe, 0x67, 0x37, 0x67, 0xce, 0x07, 0xe3, 0x6f, 0xa0, 0x4d, 0x55, 0xa1, 0x1f, 0xa5, 0x48,
0x52, 0x24, 0xac, 0x7c, 0x77, 0x6f, 0xc2, 0xd8, 0xb5, 0x93, 0x33, 0xe7, 0xc2, 0xf7, 0x0a, 0xb0, 0x58, 0xf9, 0xe2, 0xc1, 0x84, 0xb1, 0x6b, 0x37, 0x67, 0x2e, 0x84, 0xef, 0x17, 0x60, 0xfd, 0x32,
0x7a, 0x11, 0x35, 0x6b, 0x8c, 0xa1, 0x9c, 0x70, 0x70, 0x40, 0xc3, 0xd0, 0x72, 0x68, 0xe6, 0xee, 0x6a, 0xd6, 0x98, 0x40, 0x39, 0xe1, 0xe0, 0x90, 0x86, 0xa1, 0xe5, 0xd0, 0xc7, 0xee, 0x8e, 0x3a,
0xa8, 0x8b, 0xee, 0x8e, 0x3a, 0x7d, 0x6f, 0x7f, 0x40, 0xc1, 0x8e, 0x89, 0xbb, 0x07, 0xce, 0x2c, 0x7d, 0x6f, 0x3f, 0x41, 0xc1, 0x8e, 0x89, 0x7b, 0x00, 0xce, 0x3c, 0xbd, 0x66, 0x1a, 0x63, 0xfc,
0xbd, 0x66, 0x1a, 0x63, 0x3c, 0x12, 0x3b, 0x34, 0x65, 0x31, 0x7a, 0x71, 0x11, 0x8b, 0x49, 0x6d, 0x25, 0x76, 0x68, 0xca, 0x62, 0xf4, 0xe2, 0x22, 0x16, 0x93, 0xda, 0xe2, 0xfb, 0x03, 0x36, 0xc8,
0xf1, 0xfd, 0x19, 0x1b, 0x64, 0xd8, 0x80, 0xf3, 0xfc, 0x3e, 0x74, 0x11, 0x22, 0x76, 0x68, 0x96, 0xb0, 0x01, 0x17, 0xf9, 0x7d, 0xea, 0x22, 0x44, 0xec, 0xd0, 0x3c, 0xe7, 0x4f, 0x5c, 0x65, 0xe7,
0xf3, 0x07, 0xae, 0xb2, 0xfd, 0x18, 0x8a, 0xe9, 0xf6, 0x62, 0x19, 0xe0, 0xd8, 0xa3, 0x57, 0x3e, 0x6f, 0x28, 0xa6, 0xdb, 0x8b, 0x65, 0x80, 0x13, 0x8f, 0x5e, 0xfb, 0xd4, 0xe6, 0x94, 0x68, 0x39,
0xb5, 0x39, 0x25, 0x5a, 0x0e, 0xd7, 0x40, 0xed, 0x31, 0xa7, 0x7d, 0xe5, 0x86, 0x3c, 0xd4, 0x24, 0xdc, 0x00, 0xb5, 0xcf, 0x9c, 0xce, 0xb5, 0x1b, 0xf2, 0x50, 0x93, 0xb0, 0x02, 0xa5, 0x3e, 0x73,
0xac, 0x40, 0xa9, 0xc7, 0x9c, 0x3e, 0xe3, 0xfb, 0x6c, 0xe4, 0x11, 0x4d, 0x46, 0x84, 0x72, 0x9c, 0x06, 0x8c, 0x1f, 0xb0, 0xb1, 0x47, 0x34, 0x19, 0x11, 0xca, 0x71, 0xd2, 0x36, 0xf3, 0x86, 0x23,
0xb4, 0xc5, 0xbc, 0xc1, 0xd0, 0xb5, 0xb9, 0xa6, 0xa0, 0x06, 0xa5, 0x76, 0xf4, 0x06, 0x0e, 0x07, 0xd7, 0xe6, 0x9a, 0x82, 0x1a, 0x94, 0x3a, 0xd1, 0x1b, 0x38, 0x1a, 0x0e, 0x43, 0xca, 0xb5, 0xb7,
0x83, 0x90, 0x72, 0xed, 0xbd, 0xd2, 0x7c, 0x27, 0x81, 0xda, 0x4a, 0xd1, 0xe0, 0x2e, 0xe4, 0xe3, 0x4a, 0xeb, 0x8d, 0x04, 0x6a, 0x3b, 0x45, 0x83, 0x7b, 0x90, 0x8f, 0x6f, 0x18, 0x2e, 0x5b, 0xfc,
0x1b, 0x86, 0x8b, 0x16, 0x7f, 0x7a, 0xa0, 0xaa, 0x38, 0x6b, 0x3e, 0x3c, 0xc3, 0x3e, 0xa8, 0x93, 0xe9, 0x81, 0xaa, 0xe2, 0xbc, 0xf9, 0xe8, 0x1c, 0x07, 0xa0, 0x66, 0x87, 0x0c, 0xeb, 0x0b, 0x3d,
0x43, 0x86, 0xf5, 0xb9, 0x1e, 0x67, 0x6e, 0x5c, 0xf5, 0x63, 0x2c, 0x60, 0x1f, 0x8a, 0xe9, 0xad, 0xce, 0xdd, 0xb8, 0xea, 0x63, 0x2c, 0xe0, 0x00, 0x8a, 0xe9, 0xad, 0xc1, 0xed, 0x25, 0x70, 0x66,
0xc1, 0xcd, 0x05, 0x70, 0xb2, 0x57, 0xad, 0xba, 0xb1, 0xcc, 0x41, 0x9c, 0xa9, 0x2d, 0xe9, 0x27, 0xaf, 0x5a, 0x75, 0x6b, 0x95, 0x83, 0x38, 0x53, 0x0d, 0xe9, 0x1b, 0x69, 0xbf, 0xf5, 0xf2, 0xb6,
0x69, 0xaf, 0xf9, 0xfa, 0xa6, 0x26, 0x5d, 0xdf, 0xd4, 0xa4, 0xb7, 0x37, 0x35, 0xe9, 0xe5, 0x6d, 0x26, 0xdd, 0xdc, 0xd6, 0xa4, 0xd7, 0xb7, 0x35, 0xe9, 0xf9, 0x5d, 0x2d, 0x77, 0x73, 0x57, 0xcb,
0x2d, 0x77, 0x7d, 0x5b, 0xcb, 0xbd, 0xb9, 0xad, 0xe5, 0xfe, 0xd3, 0x97, 0xfd, 0x74, 0x3f, 0xc9, 0xbd, 0xba, 0xab, 0xe5, 0xfe, 0xd0, 0x57, 0xfd, 0x74, 0xff, 0x93, 0x17, 0xff, 0xbe, 0x7b, 0x17,
0x8b, 0x7f, 0xbf, 0x7c, 0x08, 0x00, 0x00, 0xff, 0xff, 0x8b, 0x1f, 0x1a, 0x5a, 0xdd, 0x07, 0x00, 0x00, 0x00, 0xff, 0xff, 0x5e, 0x57, 0x8e, 0x0b, 0xdd, 0x07, 0x00, 0x00,
0x00,
} }
func (m *Log) Marshal() (dAtA []byte, err error) { func (m *Log) Marshal() (dAtA []byte, err error) {
@ -1827,14 +1826,14 @@ func (m *LogWatchRequest) Size() (n int) {
var l int var l int
_ = l _ = l
if len(m.WatchIds) > 0 { if len(m.WatchIds) > 0 {
for _, b := range m.WatchIds { for _, s := range m.WatchIds {
l = len(b) l = len(s)
n += 1 + l + sovConsensus(uint64(l)) n += 1 + l + sovConsensus(uint64(l))
} }
} }
if len(m.UnwatchIds) > 0 { if len(m.UnwatchIds) > 0 {
for _, b := range m.UnwatchIds { for _, s := range m.UnwatchIds {
l = len(b) l = len(s)
n += 1 + l + sovConsensus(uint64(l)) n += 1 + l + sovConsensus(uint64(l))
} }
} }
@ -2041,7 +2040,7 @@ func (m *Log) Unmarshal(dAtA []byte) error {
if wireType != 2 { if wireType != 2 {
return fmt.Errorf("proto: wrong wireType = %d for field Id", wireType) return fmt.Errorf("proto: wrong wireType = %d for field Id", wireType)
} }
var byteLen int var stringLen uint64
for shift := uint(0); ; shift += 7 { for shift := uint(0); ; shift += 7 {
if shift >= 64 { if shift >= 64 {
return ErrIntOverflowConsensus return ErrIntOverflowConsensus
@ -2051,25 +2050,23 @@ func (m *Log) Unmarshal(dAtA []byte) error {
} }
b := dAtA[iNdEx] b := dAtA[iNdEx]
iNdEx++ iNdEx++
byteLen |= int(b&0x7F) << shift stringLen |= uint64(b&0x7F) << shift
if b < 0x80 { if b < 0x80 {
break break
} }
} }
if byteLen < 0 { intStringLen := int(stringLen)
if intStringLen < 0 {
return ErrInvalidLengthConsensus return ErrInvalidLengthConsensus
} }
postIndex := iNdEx + byteLen postIndex := iNdEx + intStringLen
if postIndex < 0 { if postIndex < 0 {
return ErrInvalidLengthConsensus return ErrInvalidLengthConsensus
} }
if postIndex > l { if postIndex > l {
return io.ErrUnexpectedEOF return io.ErrUnexpectedEOF
} }
m.Id = append(m.Id[:0], dAtA[iNdEx:postIndex]...) m.Id = string(dAtA[iNdEx:postIndex])
if m.Id == nil {
m.Id = []byte{}
}
iNdEx = postIndex iNdEx = postIndex
case 2: case 2:
if wireType != 2 { if wireType != 2 {
@ -2800,7 +2797,7 @@ func (m *RecordAddRequest) Unmarshal(dAtA []byte) error {
if wireType != 2 { if wireType != 2 {
return fmt.Errorf("proto: wrong wireType = %d for field LogId", wireType) return fmt.Errorf("proto: wrong wireType = %d for field LogId", wireType)
} }
var byteLen int var stringLen uint64
for shift := uint(0); ; shift += 7 { for shift := uint(0); ; shift += 7 {
if shift >= 64 { if shift >= 64 {
return ErrIntOverflowConsensus return ErrIntOverflowConsensus
@ -2810,25 +2807,23 @@ func (m *RecordAddRequest) Unmarshal(dAtA []byte) error {
} }
b := dAtA[iNdEx] b := dAtA[iNdEx]
iNdEx++ iNdEx++
byteLen |= int(b&0x7F) << shift stringLen |= uint64(b&0x7F) << shift
if b < 0x80 { if b < 0x80 {
break break
} }
} }
if byteLen < 0 { intStringLen := int(stringLen)
if intStringLen < 0 {
return ErrInvalidLengthConsensus return ErrInvalidLengthConsensus
} }
postIndex := iNdEx + byteLen postIndex := iNdEx + intStringLen
if postIndex < 0 { if postIndex < 0 {
return ErrInvalidLengthConsensus return ErrInvalidLengthConsensus
} }
if postIndex > l { if postIndex > l {
return io.ErrUnexpectedEOF return io.ErrUnexpectedEOF
} }
m.LogId = append(m.LogId[:0], dAtA[iNdEx:postIndex]...) m.LogId = string(dAtA[iNdEx:postIndex])
if m.LogId == nil {
m.LogId = []byte{}
}
iNdEx = postIndex iNdEx = postIndex
case 2: case 2:
if wireType != 2 { if wireType != 2 {
@ -2920,7 +2915,7 @@ func (m *LogWatchRequest) Unmarshal(dAtA []byte) error {
if wireType != 2 { if wireType != 2 {
return fmt.Errorf("proto: wrong wireType = %d for field WatchIds", wireType) return fmt.Errorf("proto: wrong wireType = %d for field WatchIds", wireType)
} }
var byteLen int var stringLen uint64
for shift := uint(0); ; shift += 7 { for shift := uint(0); ; shift += 7 {
if shift >= 64 { if shift >= 64 {
return ErrIntOverflowConsensus return ErrIntOverflowConsensus
@ -2930,29 +2925,29 @@ func (m *LogWatchRequest) Unmarshal(dAtA []byte) error {
} }
b := dAtA[iNdEx] b := dAtA[iNdEx]
iNdEx++ iNdEx++
byteLen |= int(b&0x7F) << shift stringLen |= uint64(b&0x7F) << shift
if b < 0x80 { if b < 0x80 {
break break
} }
} }
if byteLen < 0 { intStringLen := int(stringLen)
if intStringLen < 0 {
return ErrInvalidLengthConsensus return ErrInvalidLengthConsensus
} }
postIndex := iNdEx + byteLen postIndex := iNdEx + intStringLen
if postIndex < 0 { if postIndex < 0 {
return ErrInvalidLengthConsensus return ErrInvalidLengthConsensus
} }
if postIndex > l { if postIndex > l {
return io.ErrUnexpectedEOF return io.ErrUnexpectedEOF
} }
m.WatchIds = append(m.WatchIds, make([]byte, postIndex-iNdEx)) m.WatchIds = append(m.WatchIds, string(dAtA[iNdEx:postIndex]))
copy(m.WatchIds[len(m.WatchIds)-1], dAtA[iNdEx:postIndex])
iNdEx = postIndex iNdEx = postIndex
case 2: case 2:
if wireType != 2 { if wireType != 2 {
return fmt.Errorf("proto: wrong wireType = %d for field UnwatchIds", wireType) return fmt.Errorf("proto: wrong wireType = %d for field UnwatchIds", wireType)
} }
var byteLen int var stringLen uint64
for shift := uint(0); ; shift += 7 { for shift := uint(0); ; shift += 7 {
if shift >= 64 { if shift >= 64 {
return ErrIntOverflowConsensus return ErrIntOverflowConsensus
@ -2962,23 +2957,23 @@ func (m *LogWatchRequest) Unmarshal(dAtA []byte) error {
} }
b := dAtA[iNdEx] b := dAtA[iNdEx]
iNdEx++ iNdEx++
byteLen |= int(b&0x7F) << shift stringLen |= uint64(b&0x7F) << shift
if b < 0x80 { if b < 0x80 {
break break
} }
} }
if byteLen < 0 { intStringLen := int(stringLen)
if intStringLen < 0 {
return ErrInvalidLengthConsensus return ErrInvalidLengthConsensus
} }
postIndex := iNdEx + byteLen postIndex := iNdEx + intStringLen
if postIndex < 0 { if postIndex < 0 {
return ErrInvalidLengthConsensus return ErrInvalidLengthConsensus
} }
if postIndex > l { if postIndex > l {
return io.ErrUnexpectedEOF return io.ErrUnexpectedEOF
} }
m.UnwatchIds = append(m.UnwatchIds, make([]byte, postIndex-iNdEx)) m.UnwatchIds = append(m.UnwatchIds, string(dAtA[iNdEx:postIndex]))
copy(m.UnwatchIds[len(m.UnwatchIds)-1], dAtA[iNdEx:postIndex])
iNdEx = postIndex iNdEx = postIndex
default: default:
iNdEx = preIndex iNdEx = preIndex
@ -3034,7 +3029,7 @@ func (m *LogWatchEvent) Unmarshal(dAtA []byte) error {
if wireType != 2 { if wireType != 2 {
return fmt.Errorf("proto: wrong wireType = %d for field LogId", wireType) return fmt.Errorf("proto: wrong wireType = %d for field LogId", wireType)
} }
var byteLen int var stringLen uint64
for shift := uint(0); ; shift += 7 { for shift := uint(0); ; shift += 7 {
if shift >= 64 { if shift >= 64 {
return ErrIntOverflowConsensus return ErrIntOverflowConsensus
@ -3044,25 +3039,23 @@ func (m *LogWatchEvent) Unmarshal(dAtA []byte) error {
} }
b := dAtA[iNdEx] b := dAtA[iNdEx]
iNdEx++ iNdEx++
byteLen |= int(b&0x7F) << shift stringLen |= uint64(b&0x7F) << shift
if b < 0x80 { if b < 0x80 {
break break
} }
} }
if byteLen < 0 { intStringLen := int(stringLen)
if intStringLen < 0 {
return ErrInvalidLengthConsensus return ErrInvalidLengthConsensus
} }
postIndex := iNdEx + byteLen postIndex := iNdEx + intStringLen
if postIndex < 0 { if postIndex < 0 {
return ErrInvalidLengthConsensus return ErrInvalidLengthConsensus
} }
if postIndex > l { if postIndex > l {
return io.ErrUnexpectedEOF return io.ErrUnexpectedEOF
} }
m.LogId = append(m.LogId[:0], dAtA[iNdEx:postIndex]...) m.LogId = string(dAtA[iNdEx:postIndex])
if m.LogId == nil {
m.LogId = []byte{}
}
iNdEx = postIndex iNdEx = postIndex
case 2: case 2:
if wireType != 2 { if wireType != 2 {

10
consensus/consensusproto/protos/consensus.proto
View File

@ -13,7 +13,7 @@ enum ErrCodes {
message Log { message Log {
bytes id = 1; string id = 1;
bytes payload = 2; bytes payload = 2;
repeated RawRecordWithId records = 3; repeated RawRecordWithId records = 3;
} }
@ -57,17 +57,17 @@ message LogAddRequest {
} }
message RecordAddRequest { message RecordAddRequest {
bytes logId = 1; string logId = 1;
RawRecord record = 2; RawRecord record = 2;
} }
message LogWatchRequest { message LogWatchRequest {
repeated bytes watchIds = 1; repeated string watchIds = 1;
repeated bytes unwatchIds = 2; repeated string unwatchIds = 2;
} }
message LogWatchEvent { message LogWatchEvent {
bytes logId = 1; string logId = 1;
repeated RawRecordWithId records = 2; repeated RawRecordWithId records = 2;
Err error = 3; Err error = 3;
} }