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

cleanup and comments

Browse Source
This commit is contained in:
Sergey Cherepanov 2022-10-07 15:06:54 +03:00 committed by Mikhail Iudin
parent 7bb3d5e144
commit 77aa5a65d9
No known key found for this signature in database
GPG Key ID: FAAAA8BAABDFF1C0
10 changed files with 78 additions and 87 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
cmd/consensusnode/testclient/consensustestclient.go
View File

@ -14,7 +14,7 @@ import (
"github.com/anytypeio/go-anytype-infrastructure-experiments/config"
"github.com/anytypeio/go-anytype-infrastructure-experiments/consensus/consensusclient"
"github.com/anytypeio/go-anytype-infrastructure-experiments/consensus/consensusproto"
"github.com/anytypeio/go-anytype-infrastructure-experiments/consensus/consensusproto/consensuserrs"
"github.com/anytypeio/go-anytype-infrastructure-experiments/consensus/consensusproto/consensuserr"
"github.com/anytypeio/go-anytype-infrastructure-experiments/node/account"
"go.uber.org/zap"
"gopkg.in/mgo.v2/bson"
@ -154,7 +154,7 @@ func testCreateLogAndRecord(service consensusclient.Service) (err error) {
},
},
})
if err != consensuserrs.ErrLogExists {
if err != consensuserr.ErrLogExists {
return fmt.Errorf("unexpected error: '%v' want LogExists", zap.Error(err))
}
err = nil
@ -181,7 +181,7 @@ func testCreateLogAndRecord(service consensusclient.Service) (err error) {
PrevId: []byte(bson.NewObjectId()),
CreatedUnix: uint64(time.Now().Unix()),
})
if err != consensuserrs.ErrConflict {
if err != consensuserr.ErrConflict {
return fmt.Errorf("unexpected error: '%v' want Conflict", zap.Error(err))
}
err = nil

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

@ -59,7 +59,6 @@ func (ErrCodes) EnumDescriptor() ([]byte, []int) {
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"`
CreatedUnix uint64 `protobuf:"varint,3,opt,name=createdUnix,proto3" json:"createdUnix,omitempty"`
}
func (m *Log) Reset() { *m = Log{} }
@ -109,13 +108,6 @@ func (m *Log) GetRecords() []*Record {
return nil
}
func (m *Log) GetCreatedUnix() uint64 {
if m != nil {
return m.CreatedUnix
}
return 0
}
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"`
@ -436,37 +428,37 @@ func init() {
}
var fileDescriptor_b8d7f1c16b400059 = []byte{
// 480 bytes of a gzipped FileDescriptorProto
0x1f, 0x8b, 0x08, 0x00, 0x00, 0x00, 0x00, 0x00, 0x02, 0xff, 0x94, 0x53, 0x41, 0x8b, 0xd3, 0x40,
0x14, 0xee, 0x24, 0x6b, 0xb7, 0x7d, 0x69, 0xbb, 0xf1, 0xb1, 0x48, 0xe8, 0x62, 0x08, 0xf1, 0x12,
0x44, 0xba, 0x52, 0x05, 0x4f, 0x1e, 0xd6, 0x52, 0xa1, 0x52, 0x2d, 0x04, 0xaa, 0xe0, 0xc9, 0x98,
0x99, 0xc6, 0xb0, 0x21, 0x53, 0x67, 0xd2, 0xb5, 0xfb, 0x2f, 0xfc, 0x21, 0xfe, 0x10, 0x6f, 0xee,
0xd1, 0xa3, 0xb4, 0x7f, 0x44, 0x3a, 0x69, 0x6a, 0xb2, 0xdd, 0x1e, 0xbc, 0x24, 0x79, 0xdf, 0x37,
0xef, 0x7d, 0xdf, 0xbc, 0x8f, 0xc0, 0x79, 0xc8, 0x53, 0xc9, 0x52, 0xb9, 0x90, 0xff, 0xbe, 0xe6,
0x82, 0x67, 0xfc, 0x5c, 0x3d, 0x4b, 0x68, 0x4f, 0x01, 0xd8, 0x0a, 0xd2, 0xeb, 0x41, 0x81, 0xb9,
0x11, 0xe8, 0x63, 0x1e, 0x61, 0x07, 0xb4, 0x98, 0x5a, 0xc4, 0x21, 0x5e, 0xcb, 0xd7, 0x62, 0x8a,
0x3d, 0x38, 0x16, 0x2c, 0xe4, 0x82, 0x4a, 0x4b, 0x73, 0x74, 0xcf, 0xe8, 0x9f, 0xf6, 0xca, 0x6d,
0x3d, 0x5f, 0x91, 0x7e, 0x71, 0x08, 0x1d, 0x30, 0x42, 0xc1, 0x82, 0x8c, 0xd1, 0x69, 0x1a, 0x2f,
0x2d, 0xdd, 0x21, 0xde, 0x91, 0x5f, 0x86, 0xdc, 0x04, 0xea, 0x79, 0xd3, 0x9e, 0xd6, 0x03, 0xa8,
0xcf, 0x05, 0xbb, 0x1a, 0x51, 0x4b, 0x53, 0xd8, 0xb6, 0x42, 0x0b, 0x8e, 0xe7, 0xc1, 0x75, 0xc2,
0x03, 0xaa, 0xe6, 0xb5, 0xfc, 0xa2, 0xbc, 0xad, 0x76, 0xb4, 0xaf, 0x76, 0x04, 0xda, 0xe4, 0xd2,
0x7d, 0x0e, 0xed, 0x0b, 0x4a, 0xc7, 0x3c, 0xf2, 0xd9, 0xd7, 0x05, 0x93, 0x19, 0x3e, 0x02, 0x3d,
0xe1, 0x91, 0xd2, 0x36, 0xfa, 0xf7, 0xab, 0x57, 0xda, 0x1c, 0xdb, 0xb0, 0xee, 0x7b, 0x30, 0x2f,
0x28, 0xdd, 0xde, 0x70, 0xdb, 0x78, 0x0a, 0xf7, 0x12, 0x1e, 0x8d, 0x0a, 0xdb, 0x79, 0x81, 0x4f,
0xa0, 0x9e, 0x2f, 0x40, 0x39, 0x3f, 0xb4, 0xa4, 0xed, 0x19, 0xf7, 0x2d, 0x9c, 0x7c, 0x08, 0xb2,
0xf0, 0x4b, 0xc9, 0x4f, 0x17, 0x1a, 0xdf, 0x36, 0xd0, 0x88, 0x4a, 0x8b, 0x38, 0xba, 0xd7, 0xf2,
0x77, 0x35, 0xda, 0x00, 0x8b, 0x74, 0xc7, 0x6a, 0x8a, 0x2d, 0x21, 0xee, 0x14, 0xda, 0xc5, 0xb8,
0xe1, 0x15, 0x4b, 0x0f, 0x79, 0xfc, 0xcf, 0x24, 0x1f, 0x7f, 0x82, 0xc6, 0x50, 0x88, 0x01, 0xa7,
0x4c, 0x62, 0x07, 0x60, 0x9a, 0xb2, 0xe5, 0x9c, 0x85, 0x19, 0xa3, 0x66, 0x0d, 0xdb, 0xd0, 0xdc,
0xa8, 0x2d, 0x63, 0x99, 0x49, 0x93, 0xe0, 0x09, 0x18, 0x63, 0x1e, 0xbd, 0xe3, 0xd9, 0x6b, 0xbe,
0x48, 0xa9, 0xa9, 0x21, 0x42, 0x27, 0x1f, 0x37, 0xe0, 0xe9, 0x2c, 0x89, 0xc3, 0xcc, 0xd4, 0xd1,
0x04, 0x63, 0x28, 0x04, 0x17, 0x93, 0xd9, 0x4c, 0xb2, 0xcc, 0xfc, 0xa1, 0xf5, 0x7f, 0x11, 0x68,
0xee, 0xf4, 0xf1, 0x05, 0xd4, 0xf3, 0x8c, 0xf0, 0xac, 0x6a, 0xac, 0x92, 0x5c, 0xd7, 0xac, 0x92,
0x93, 0x4b, 0x7c, 0x09, 0xcd, 0x5d, 0x4c, 0x68, 0xef, 0xf5, 0x56, 0xf2, 0xbb, 0xa3, 0xfd, 0x0d,
0x34, 0x8a, 0xf5, 0xe1, 0xc3, 0x2a, 0x7b, 0x2b, 0xa5, 0xee, 0xd9, 0xdd, 0xb4, 0xda, 0xba, 0x47,
0x9e, 0x92, 0x57, 0xfd, 0x9f, 0x2b, 0x9b, 0xdc, 0xac, 0x6c, 0xf2, 0x67, 0x65, 0x93, 0xef, 0x6b,
0xbb, 0x76, 0xb3, 0xb6, 0x6b, 0xbf, 0xd7, 0x76, 0xed, 0xa3, 0x75, 0xe8, 0xef, 0xfc, 0x5c, 0x57,
0xaf, 0x67, 0x7f, 0x03, 0x00, 0x00, 0xff, 0xff, 0xf0, 0xc9, 0xa1, 0x1f, 0xc0, 0x03, 0x00, 0x00,
// 474 bytes of a gzipped FileDescriptorProto
0x1f, 0x8b, 0x08, 0x00, 0x00, 0x00, 0x00, 0x00, 0x02, 0xff, 0x94, 0x53, 0x4f, 0x8b, 0xd3, 0x40,
0x14, 0xef, 0x24, 0xb5, 0xdb, 0xbe, 0xb4, 0xdd, 0xf8, 0x58, 0x24, 0x74, 0x31, 0x84, 0x78, 0x09,
0x22, 0x5d, 0xa9, 0x82, 0x27, 0x0f, 0x6b, 0xa9, 0x50, 0xa9, 0x16, 0x02, 0x55, 0xf0, 0x64, 0xcc,
0x4c, 0x63, 0xd8, 0x90, 0xa9, 0x33, 0xe9, 0xda, 0xfd, 0x16, 0x7e, 0x10, 0x3f, 0x88, 0x37, 0xf7,
0xe8, 0x51, 0xda, 0x2f, 0x22, 0x9d, 0x34, 0x35, 0x71, 0xb7, 0x87, 0xbd, 0x24, 0xf3, 0x7e, 0xbf,
0xf7, 0xe7, 0x37, 0xef, 0xc7, 0xc0, 0x59, 0xc8, 0x53, 0xc9, 0x52, 0xb9, 0x94, 0xff, 0x4e, 0x0b,
0xc1, 0x33, 0x7e, 0xa6, 0xbe, 0x25, 0xb4, 0xaf, 0x00, 0x6c, 0x07, 0xe9, 0xd5, 0xb0, 0xc0, 0xdc,
0x11, 0xe8, 0x13, 0x1e, 0x61, 0x17, 0xb4, 0x98, 0x5a, 0xc4, 0x21, 0x5e, 0xdb, 0xd7, 0x62, 0x8a,
0x7d, 0x38, 0x12, 0x2c, 0xe4, 0x82, 0x4a, 0x4b, 0x73, 0x74, 0xcf, 0x18, 0x9c, 0xf4, 0xcb, 0x65,
0x7d, 0x5f, 0x91, 0x7e, 0x91, 0xe4, 0x26, 0xd0, 0xc8, 0xa1, 0x1b, 0x9d, 0x1e, 0x40, 0x63, 0x21,
0xd8, 0xe5, 0x98, 0x5a, 0x9a, 0xc2, 0x76, 0x11, 0x5a, 0x70, 0xb4, 0x08, 0xae, 0x12, 0x1e, 0x50,
0x4b, 0x57, 0x44, 0x11, 0xa2, 0x03, 0x46, 0x28, 0x58, 0x90, 0x31, 0x3a, 0x4b, 0xe3, 0x95, 0x55,
0x77, 0x88, 0x57, 0xf7, 0xcb, 0x90, 0x5b, 0x07, 0x6d, 0x7a, 0xe1, 0x3e, 0x87, 0xce, 0x39, 0xa5,
0x13, 0x1e, 0xf9, 0xec, 0xeb, 0x92, 0xc9, 0x0c, 0x1f, 0x81, 0x9e, 0xf0, 0x48, 0xcd, 0x36, 0x06,
0xf7, 0xab, 0x82, 0xb7, 0x69, 0x5b, 0xd6, 0x7d, 0x0f, 0xe6, 0x39, 0xa5, 0x3b, 0xfd, 0xbb, 0xc2,
0x13, 0xb8, 0x97, 0xf0, 0x68, 0x5c, 0xc8, 0xce, 0x03, 0x7c, 0x02, 0x8d, 0xfc, 0x7a, 0x4a, 0xf9,
0xa1, 0x15, 0xec, 0x72, 0xdc, 0xb7, 0x70, 0xfc, 0x21, 0xc8, 0xc2, 0x2f, 0x25, 0x3d, 0x3d, 0x68,
0x7e, 0xdb, 0x42, 0x63, 0x2a, 0x2d, 0xe2, 0xe8, 0x5e, 0xdb, 0xdf, 0xc7, 0x68, 0x03, 0x2c, 0xd3,
0x3d, 0xab, 0x29, 0xb6, 0x84, 0xb8, 0x33, 0xe8, 0x14, 0xed, 0x46, 0x97, 0x2c, 0x3d, 0xa4, 0xf1,
0x8e, 0x3e, 0x3d, 0xfe, 0x04, 0xcd, 0x91, 0x10, 0x43, 0x4e, 0x99, 0xc4, 0x2e, 0xc0, 0x2c, 0x65,
0xab, 0x05, 0x0b, 0x33, 0x46, 0xcd, 0x1a, 0x76, 0xa0, 0xb5, 0x9d, 0xb6, 0x8a, 0x65, 0x26, 0x4d,
0x82, 0xc7, 0x60, 0x4c, 0x78, 0xf4, 0x8e, 0x67, 0xaf, 0xf9, 0x32, 0xa5, 0xa6, 0x86, 0x08, 0xdd,
0xbc, 0xdd, 0x90, 0xa7, 0xf3, 0x24, 0x0e, 0x33, 0x53, 0x47, 0x13, 0x8c, 0x91, 0x10, 0x5c, 0x4c,
0xe7, 0x73, 0xc9, 0x32, 0xf3, 0x87, 0x36, 0xf8, 0x45, 0xa0, 0xb5, 0x9f, 0x8f, 0x2f, 0xa0, 0x91,
0x7b, 0x84, 0xa7, 0x55, 0x61, 0x15, 0xe7, 0x7a, 0x66, 0x95, 0x9c, 0x5e, 0xe0, 0x4b, 0x68, 0xed,
0x6d, 0x42, 0xfb, 0x46, 0x6d, 0xc5, 0xbf, 0x5b, 0xca, 0xdf, 0x40, 0xb3, 0x58, 0x1f, 0x3e, 0xac,
0xb2, 0xff, 0xb9, 0xd4, 0x3b, 0xbd, 0x9d, 0x56, 0x5b, 0xf7, 0xc8, 0x53, 0xf2, 0x6a, 0xf0, 0x73,
0x6d, 0x93, 0xeb, 0xb5, 0x4d, 0xfe, 0xac, 0x6d, 0xf2, 0x7d, 0x63, 0xd7, 0xae, 0x37, 0x76, 0xed,
0xf7, 0xc6, 0xae, 0x7d, 0xb4, 0x0e, 0xbd, 0xbd, 0xcf, 0x0d, 0xf5, 0x7b, 0xf6, 0x37, 0x00, 0x00,
0xff, 0xff, 0x9f, 0x6e, 0xb2, 0xf8, 0x9e, 0x03, 0x00, 0x00,
}
func (m *Log) Marshal() (dAtA []byte, err error) {
@ -489,11 +481,6 @@ func (m *Log) MarshalToSizedBuffer(dAtA []byte) (int, error) {
_ = i
var l int
_ = l
if m.CreatedUnix != 0 {
i = encodeVarintConsensus(dAtA, i, uint64(m.CreatedUnix))
i--
dAtA[i] = 0x18
}
if len(m.Records) > 0 {
for iNdEx := len(m.Records) - 1; iNdEx >= 0; iNdEx-- {
{
@ -779,9 +766,6 @@ func (m *Log) Size() (n int) {
n += 1 + l + sovConsensus(uint64(l))
}
}
if m.CreatedUnix != 0 {
n += 1 + sovConsensus(uint64(m.CreatedUnix))
}
return n
}
@ -991,25 +975,6 @@ func (m *Log) Unmarshal(dAtA []byte) error {
return err
}
iNdEx = postIndex
case 3:
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:])

2
consensus/consensusproto/consensuserrs/errors.go → consensus/consensusproto/consensuserr/errors.go
View File

@ -1,4 +1,4 @@
package consensuserrs
package consensuserr
import (
"fmt"

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

@ -15,7 +15,6 @@ enum ErrCodes {
message Log {
bytes id = 1;
repeated Record records = 2;
uint64 createdUnix = 3;
}
message Record {

1
consensus/consensusrpc/consensrpc.go
View File

@ -17,6 +17,7 @@ func New() app.Component {
return &consensusRpc{}
}
// consensusRpc implements consensus rpc server
type consensusRpc struct {
db db.Service
stream stream.Service

19
consensus/db/db.go
View File

@ -7,7 +7,7 @@ import (
"github.com/anytypeio/go-anytype-infrastructure-experiments/app/logger"
"github.com/anytypeio/go-anytype-infrastructure-experiments/consensus"
"github.com/anytypeio/go-anytype-infrastructure-experiments/consensus/config"
"github.com/anytypeio/go-anytype-infrastructure-experiments/consensus/consensusproto/consensuserrs"
"github.com/anytypeio/go-anytype-infrastructure-experiments/consensus/consensusproto/consensuserr"
"go.mongodb.org/mongo-driver/mongo"
"go.mongodb.org/mongo-driver/mongo/options"
"go.uber.org/zap"
@ -24,9 +24,14 @@ func New() Service {
type ChangeReceiver func(logId []byte, records []consensus.Record)
type Service interface {
// AddLog adds new log db
AddLog(ctx context.Context, log consensus.Log) (err error)
// AddRecord adds new record to existing log
// returns consensuserr.ErrConflict if record didn't match or log not found
AddRecord(ctx context.Context, logId []byte, record consensus.Record) (err error)
// FetchLog gets log by id
FetchLog(ctx context.Context, logId []byte) (log consensus.Log, err error)
// SetChangeReceiver sets the receiver for updates, it must be called before app.Run stage
SetChangeReceiver(receiver ChangeReceiver) (err error)
app.ComponentRunnable
}
@ -69,7 +74,7 @@ func (s *service) Run(ctx context.Context) (err error) {
func (s *service) AddLog(ctx context.Context, l consensus.Log) (err error) {
_, err = s.logColl.InsertOne(ctx, l)
if mongo.IsDuplicateKeyError(err) {
return consensuserrs.ErrLogExists
return consensuserr.ErrLogExists
}
return
}
@ -101,10 +106,10 @@ func (s *service) AddRecord(ctx context.Context, logId []byte, record consensus.
}, upd)
if err != nil {
log.Error("addRecord update error", zap.Error(err))
return consensuserrs.ErrUnexpected
return consensuserr.ErrUnexpected
}
if result.ModifiedCount == 0 {
return consensuserrs.ErrConflict
return consensuserr.ErrConflict
}
return
}
@ -112,7 +117,7 @@ func (s *service) AddRecord(ctx context.Context, logId []byte, record consensus.
func (s *service) FetchLog(ctx context.Context, logId []byte) (l consensus.Log, err error) {
if err = s.logColl.FindOne(ctx, findLogQuery{Id: logId}).Decode(&l); err != nil {
if err == mongo.ErrNoDocuments {
err = consensuserrs.ErrLogNotFound
err = consensuserr.ErrLogNotFound
}
return
}
@ -162,7 +167,9 @@ func (s *service) streamListener(stream *mongo.ChangeStream) {
for stream.Next(s.streamCtx) {
var res streamResult
if err := stream.Decode(&res); err != nil {
log.Error("stream decode error:", zap.Error(err))
// mongo driver maintains connections and handles reconnects so that the stream will work as usual in these cases
// here we have an unexpected error and should stop any operations to avoid an inconsistent state between db and cache
log.Fatal("stream decode error:", zap.Error(err))
}
s.changeReceiver(res.DocumentKey.Id, res.UpdateDescription.UpdateFields.Records)
}

4
consensus/db/db_test.go
View File

@ -5,7 +5,7 @@ import (
"github.com/anytypeio/go-anytype-infrastructure-experiments/app"
"github.com/anytypeio/go-anytype-infrastructure-experiments/consensus"
"github.com/anytypeio/go-anytype-infrastructure-experiments/consensus/config"
"github.com/anytypeio/go-anytype-infrastructure-experiments/consensus/consensusproto/consensuserrs"
"github.com/anytypeio/go-anytype-infrastructure-experiments/consensus/consensusproto/consensuserr"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
"testing"
@ -100,7 +100,7 @@ func TestService_FetchLog(t *testing.T) {
defer fx.Finish(t)
l, err := fx.FetchLog(ctx, []byte("not exists"))
assert.Empty(t, l)
assert.ErrorIs(t, err, consensuserrs.ErrLogNotFound)
assert.ErrorIs(t, err, consensuserr.ErrLogNotFound)
})
}

19
consensus/stream/object.go
View File

@ -5,6 +5,7 @@ import (
"sync"
)
// object is a cache entry that holds the actual log state and maintains added streams
type object struct {
logId []byte
records []consensus.Record
@ -14,6 +15,8 @@ type object struct {
mu sync.Mutex
}
// AddRecords adds new records to the log and sends new records to streams
// The records source is db and called via stream.Service
func (o *object) AddRecords(recs []consensus.Record) {
o.mu.Lock()
defer o.mu.Unlock()
@ -28,12 +31,14 @@ func (o *object) AddRecords(recs []consensus.Record) {
}
}
// Records returns all log records
func (o *object) Records() []consensus.Record {
o.mu.Lock()
defer o.mu.Unlock()
return o.records
}
// AddStream adds stream to the object
func (o *object) AddStream(s *Stream) {
o.mu.Lock()
defer o.mu.Unlock()
@ -42,18 +47,20 @@ func (o *object) AddStream(s *Stream) {
return
}
func (o *object) Locked() bool {
o.mu.Lock()
defer o.mu.Unlock()
return len(o.streams) > 0
}
// RemoveStream remove stream from object
func (o *object) RemoveStream(id uint64) {
o.mu.Lock()
defer o.mu.Unlock()
delete(o.streams, id)
}
// Locked indicates that object can be garbage collected
func (o *object) Locked() bool {
o.mu.Lock()
defer o.mu.Unlock()
return len(o.streams) > 0
}
func (o *object) Close() (err error) {
return nil
}

4
consensus/stream/service.go
View File

@ -32,7 +32,9 @@ func New() Service {
return &service{}
}
// Service maintains a cache for logs (receive updates from db) and creates new stream objects with able to subscribe/unsubscribe to log ids
type Service interface {
// NewStream creates new stream with able to watch and unwatch log ids
NewStream() *Stream
app.ComponentRunnable
}
@ -71,6 +73,7 @@ func (s *service) NewStream() *Stream {
}
}
// AddStream to object with given logId
func (s *service) AddStream(ctx context.Context, logId []byte, stream *Stream) (err error) {
obj, err := s.getObject(ctx, logId)
if err != nil {
@ -80,6 +83,7 @@ func (s *service) AddStream(ctx context.Context, logId []byte, stream *Stream) (
return
}
// RemoveStream from object with five logId
func (s *service) RemoveStream(ctx context.Context, logId []byte, streamId uint64) (err error) {
obj, err := s.getObject(ctx, logId)
if err != nil {

8
consensus/stream/stream.go
View File

@ -8,6 +8,7 @@ import (
"sync"
)
// Stream is a buffer that receives updates from object and gives back to a client
type Stream struct {
id uint64
logIds map[string]struct{}
@ -16,6 +17,7 @@ type Stream struct {
s *service
}
// LogIds returns watched log ids
func (s *Stream) LogIds() [][]byte {
s.mu.Lock()
defer s.mu.Unlock()
@ -26,14 +28,18 @@ func (s *Stream) LogIds() [][]byte {
return logIds
}
// AddRecords adds new records to stream, called by objects
func (s *Stream) AddRecords(logId []byte, records []consensus.Record) (err error) {
return s.mb.Add(consensus.Log{Id: logId, Records: records})
}
// WaitLogs wait for new log records
// empty returned slice means that stream is closed
func (s *Stream) WaitLogs() []consensus.Log {
return s.mb.Wait()
}
// WatchIds adds given ids to subscription
func (s *Stream) WatchIds(ctx context.Context, logIds [][]byte) {
s.mu.Lock()
defer s.mu.Unlock()
@ -49,6 +55,7 @@ func (s *Stream) WatchIds(ctx context.Context, logIds [][]byte) {
return
}
// UnwatchIds removes given ids from subscription
func (s *Stream) UnwatchIds(ctx context.Context, logIds [][]byte) {
s.mu.Lock()
defer s.mu.Unlock()
@ -64,6 +71,7 @@ func (s *Stream) UnwatchIds(ctx context.Context, logIds [][]byte) {
return
}
// Close closes stream and unsubscribes all ids
func (s *Stream) Close() {
_ = s.mb.Close()
s.mu.Lock()