From 5b27a186988a725acfe5880db910f61169aa3aea Mon Sep 17 00:00:00 2001 From: Injun Song Date: Wed, 17 May 2023 01:38:05 +0900 Subject: [PATCH] feat(storagenode): change append rpc from unary to stream This PR changes the Append RPC lifestyle from a unary to a bidirectional stream. However, it neither adds nor updates stream-styled Append. Therefore, this change is transparent for users. This is starting point to support streaming Append API, and LocalOrderClient mentioned on #433. --- internal/storagenode/client/log_client.go | 20 +- .../storagenode/client/log_client_test.go | 18 +- internal/storagenode/client/testing.go | 20 ++ internal/storagenode/log_server.go | 62 +++-- internal/storagenode/storagenode_test.go | 74 ++---- internal/varlogcli/stress.go | 67 ------ proto/snpb/generate.go | 2 +- proto/snpb/log_io.pb.go | 214 ++++++++++-------- proto/snpb/log_io.proto | 9 +- proto/snpb/mock/snpb_mock.go | 164 ++++++++++++-- 10 files changed, 391 insertions(+), 259 deletions(-) create mode 100644 internal/storagenode/client/testing.go delete mode 100644 internal/varlogcli/stress.go diff --git a/internal/storagenode/client/log_client.go b/internal/storagenode/client/log_client.go index 331165c6d..6b90155bb 100644 --- a/internal/storagenode/client/log_client.go +++ b/internal/storagenode/client/log_client.go @@ -42,14 +42,30 @@ func (c *LogClient) reset(rpcConn *rpc.Conn, _ types.ClusterID, target varlogpb. // The backup indicates the storage nodes that have backup replicas of that log stream. // It returns valid GLSN if the append completes successfully. func (c *LogClient) Append(ctx context.Context, tpid types.TopicID, lsid types.LogStreamID, data [][]byte) ([]snpb.AppendResult, error) { + ctx, cancel := context.WithCancel(ctx) + defer cancel() + + stream, err := c.rpcClient.Append(ctx) + if err != nil { + return nil, err + } + defer func() { + _ = stream.CloseSend() + }() + req := &snpb.AppendRequest{ TopicID: tpid, LogStreamID: lsid, Payload: data, } - rsp, err := c.rpcClient.Append(ctx, req) + err = stream.Send(req) if err != nil { - return nil, fmt.Errorf("logclient: %w", verrors.FromStatusError(err)) + return nil, err + } + + rsp, err := stream.Recv() + if err != nil { + return nil, err } return rsp.Results, nil } diff --git a/internal/storagenode/client/log_client_test.go b/internal/storagenode/client/log_client_test.go index 0aca66b26..4143640ad 100644 --- a/internal/storagenode/client/log_client_test.go +++ b/internal/storagenode/client/log_client_test.go @@ -30,6 +30,7 @@ type storageNode struct { llsn types.LLSN logEntries map[types.GLSN][]byte glsnToLLSN map[types.GLSN]types.LLSN + appendReq *snpb.AppendRequest mu sync.Mutex } @@ -46,16 +47,19 @@ func newMockStorageNodeServiceClient(ctrl *gomock.Controller, sn *storageNode, t mockClient := mock.NewMockLogIOClient(ctrl) // Append - mockClient.EXPECT().Append( - gomock.Any(), - gomock.Any(), - gomock.Any(), - ).DoAndReturn(func(ctx context.Context, req *snpb.AppendRequest, opts ...grpc.CallOption) (*snpb.AppendResponse, error) { + mockAppendClient := mock.NewMockLogIO_AppendClient(ctrl) + mockAppendClient.EXPECT().Send(gomock.Any()).DoAndReturn(func(req *snpb.AppendRequest) error { + sn.mu.Lock() + defer sn.mu.Unlock() + sn.appendReq = req + return nil + }).AnyTimes() + mockAppendClient.EXPECT().Recv().DoAndReturn(func() (*snpb.AppendResponse, error) { sn.mu.Lock() defer sn.mu.Unlock() rsp := &snpb.AppendResponse{} - for _, buf := range req.GetPayload() { + for _, buf := range sn.appendReq.GetPayload() { sn.logEntries[sn.glsn] = buf sn.glsnToLLSN[sn.glsn] = sn.llsn rsp.Results = append(rsp.Results, snpb.AppendResult{ @@ -71,6 +75,8 @@ func newMockStorageNodeServiceClient(ctrl *gomock.Controller, sn *storageNode, t } return rsp, nil }).AnyTimes() + mockAppendClient.EXPECT().CloseSend().Return(nil).AnyTimes() + mockClient.EXPECT().Append(gomock.Any(), gomock.Any()).Return(mockAppendClient, nil).AnyTimes() // Read mockClient.EXPECT().Read( diff --git a/internal/storagenode/client/testing.go b/internal/storagenode/client/testing.go new file mode 100644 index 000000000..c03b489e5 --- /dev/null +++ b/internal/storagenode/client/testing.go @@ -0,0 +1,20 @@ +package client + +import ( + "testing" + + "github.com/stretchr/testify/require" + + "github.com/kakao/varlog/proto/snpb" + "github.com/kakao/varlog/proto/varlogpb" +) + +func TestNewLogClient(t *testing.T, rpcClient snpb.LogIOClient, target varlogpb.StorageNode) *LogClient { + require.NotNil(t, rpcClient) + require.False(t, target.StorageNodeID.Invalid()) + require.NotEmpty(t, target.Address) + return &LogClient{ + rpcClient: rpcClient, + target: target, + } +} diff --git a/internal/storagenode/log_server.go b/internal/storagenode/log_server.go index cd8f7471b..da1113f32 100644 --- a/internal/storagenode/log_server.go +++ b/internal/storagenode/log_server.go @@ -3,6 +3,7 @@ package storagenode import ( "context" "errors" + "io" pbtypes "github.com/gogo/protobuf/types" "go.uber.org/multierr" @@ -22,33 +23,48 @@ type logServer struct { var _ snpb.LogIOServer = (*logServer)(nil) -func (ls logServer) Append(ctx context.Context, req *snpb.AppendRequest) (*snpb.AppendResponse, error) { - err := snpb.ValidateTopicLogStream(req) - if err != nil { - return nil, status.Error(codes.InvalidArgument, err.Error()) - } +func (ls logServer) Append(stream snpb.LogIO_AppendServer) (err error) { + req, rsp := &snpb.AppendRequest{}, &snpb.AppendResponse{} - payload := req.GetPayload() - req.Payload = nil - lse, loaded := ls.sn.executors.Load(req.TopicID, req.LogStreamID) - if !loaded { - return nil, status.Error(codes.NotFound, "no such log stream") - } + for { + req.Reset() + err = stream.RecvMsg(req) + if err == io.EOF { + return nil + } + if err != nil { + return err + } - res, err := lse.Append(ctx, payload) - if err != nil { - var code codes.Code - switch err { - case verrors.ErrSealed: - code = codes.FailedPrecondition - case snerrors.ErrNotPrimary: - code = codes.Unavailable - default: - code = status.FromContextError(err).Code() + err = snpb.ValidateTopicLogStream(req) + if err != nil { + return status.Error(codes.InvalidArgument, err.Error()) + } + + lse, loaded := ls.sn.executors.Load(req.TopicID, req.LogStreamID) + if !loaded { + return status.Error(codes.NotFound, "no such log stream") + } + res, err := lse.Append(stream.Context(), req.Payload) + if err != nil { + var code codes.Code + switch err { + case verrors.ErrSealed: + code = codes.FailedPrecondition + case snerrors.ErrNotPrimary: + code = codes.Unavailable + default: + code = status.FromContextError(err).Code() + } + return status.Error(code, err.Error()) + } + + rsp.Results = res + err = stream.Send(rsp) + if err != nil { + return err } - return nil, status.Error(code, err.Error()) } - return &snpb.AppendResponse{Results: res}, nil } func (ls logServer) Read(context.Context, *snpb.ReadRequest) (*snpb.ReadResponse, error) { diff --git a/internal/storagenode/storagenode_test.go b/internal/storagenode/storagenode_test.go index cddf2e45b..6cbcd3f9c 100644 --- a/internal/storagenode/storagenode_test.go +++ b/internal/storagenode/storagenode_test.go @@ -456,61 +456,45 @@ func TestStorageNode_Append(t *testing.T) { tcs := []struct { name string - testf func(t *testing.T, addr string, lc snpb.LogIOClient) + testf func(t *testing.T, addr string, lc *client.LogClient) }{ { name: "InvalidTopicID", - testf: func(t *testing.T, _ string, lc snpb.LogIOClient) { + testf: func(t *testing.T, _ string, lc *client.LogClient) { const invalidTopicID = types.TopicID(0) - _, err := lc.Append(context.Background(), &snpb.AppendRequest{ - TopicID: invalidTopicID, - LogStreamID: lsid, - Payload: payload, - }) + _, err := lc.Append(context.Background(), invalidTopicID, lsid, payload) require.Error(t, err) require.Equal(t, codes.InvalidArgument, status.Code(err)) }, }, { name: "InvalidLogStreamID", - testf: func(t *testing.T, _ string, lc snpb.LogIOClient) { + testf: func(t *testing.T, _ string, lc *client.LogClient) { const invalidLogStreamID = types.LogStreamID(0) - _, err := lc.Append(context.Background(), &snpb.AppendRequest{ - TopicID: tpid, - LogStreamID: invalidLogStreamID, - Payload: payload, - }) + _, err := lc.Append(context.Background(), tpid, invalidLogStreamID, payload) require.Error(t, err) require.Equal(t, codes.InvalidArgument, status.Code(err)) }, }, { name: "NoSuchTopic", - testf: func(t *testing.T, _ string, lc snpb.LogIOClient) { - _, err := lc.Append(context.Background(), &snpb.AppendRequest{ - TopicID: tpid + 1, - LogStreamID: lsid, - Payload: payload, - }) + testf: func(t *testing.T, _ string, lc *client.LogClient) { + _, err := lc.Append(context.Background(), tpid+1, lsid, payload) require.Error(t, err) require.Equal(t, codes.NotFound, status.Code(err)) }, }, { name: "NoSuchLogStream", - testf: func(t *testing.T, _ string, lc snpb.LogIOClient) { - _, err := lc.Append(context.Background(), &snpb.AppendRequest{ - TopicID: tpid, - LogStreamID: lsid + 1, - Payload: payload, - }) + testf: func(t *testing.T, _ string, lc *client.LogClient) { + _, err := lc.Append(context.Background(), tpid, lsid+1, payload) require.Error(t, err) require.Equal(t, codes.NotFound, status.Code(err)) }, }, { name: "NotPrimary", - testf: func(t *testing.T, addr string, lc snpb.LogIOClient) { + testf: func(t *testing.T, addr string, lc *client.LogClient) { lss, lastGLSN := TestSealLogStreamReplica(t, cid, snid, tpid, lsid, types.InvalidGLSN, addr) require.Equal(t, varlogpb.LogStreamStatusSealed, lss) require.True(t, lastGLSN.Invalid()) @@ -538,34 +522,26 @@ func TestStorageNode_Append(t *testing.T) { }, }, addr) - _, err := lc.Append(context.Background(), &snpb.AppendRequest{ - TopicID: tpid, - LogStreamID: lsid, - Payload: payload, - }) + _, err := lc.Append(context.Background(), tpid, lsid, payload) require.Error(t, err) require.Equal(t, codes.Unavailable, status.Code(err)) }, }, { name: "Sealed", - testf: func(t *testing.T, addr string, lc snpb.LogIOClient) { + testf: func(t *testing.T, addr string, lc *client.LogClient) { lss, lastGLSN := TestSealLogStreamReplica(t, cid, snid, tpid, lsid, types.InvalidGLSN, addr) require.Equal(t, varlogpb.LogStreamStatusSealed, lss) require.True(t, lastGLSN.Invalid()) - _, err := lc.Append(context.Background(), &snpb.AppendRequest{ - TopicID: tpid, - LogStreamID: lsid, - Payload: payload, - }) + _, err := lc.Append(context.Background(), tpid, lsid, payload) require.Error(t, err) require.Equal(t, codes.FailedPrecondition, status.Code(err)) }, }, { name: "DeadlineExceeded", - testf: func(t *testing.T, addr string, lc snpb.LogIOClient) { + testf: func(t *testing.T, addr string, lc *client.LogClient) { lss, lastGLSN := TestSealLogStreamReplica(t, cid, snid, tpid, lsid, types.InvalidGLSN, addr) require.Equal(t, varlogpb.LogStreamStatusSealed, lss) require.True(t, lastGLSN.Invalid()) @@ -585,18 +561,14 @@ func TestStorageNode_Append(t *testing.T) { ctx, cancel := context.WithTimeout(context.Background(), 10*time.Millisecond) defer cancel() - _, err := lc.Append(ctx, &snpb.AppendRequest{ - TopicID: tpid, - LogStreamID: lsid, - Payload: payload, - }) + _, err := lc.Append(ctx, tpid, lsid, payload) require.Error(t, err) require.Equal(t, codes.DeadlineExceeded, status.Code(err)) }, }, { name: "Canceled", - testf: func(t *testing.T, addr string, lc snpb.LogIOClient) { + testf: func(t *testing.T, addr string, lc *client.LogClient) { lss, lastGLSN := TestSealLogStreamReplica(t, cid, snid, tpid, lsid, types.InvalidGLSN, addr) require.Equal(t, varlogpb.LogStreamStatusSealed, lss) require.True(t, lastGLSN.Invalid()) @@ -623,11 +595,7 @@ func TestStorageNode_Append(t *testing.T) { wg.Add(1) go func() { defer wg.Done() - _, err := lc.Append(ctx, &snpb.AppendRequest{ - TopicID: tpid, - LogStreamID: lsid, - Payload: payload, - }) + _, err := lc.Append(ctx, tpid, lsid, payload) assert.Error(t, err) assert.Equal(t, codes.Canceled, status.Code(err)) }() @@ -663,8 +631,12 @@ func TestStorageNode_Append(t *testing.T) { defer func() { require.NoError(t, rpcConn.Close()) }() - lc := snpb.NewLogIOClient(rpcConn.Conn) - + lc := client.TestNewLogClient(t, snpb.NewLogIOClient(rpcConn.Conn), + varlogpb.StorageNode{ + StorageNodeID: snid, + Address: addr, + }, + ) tc.testf(t, addr, lc) }) } diff --git a/internal/varlogcli/stress.go b/internal/varlogcli/stress.go deleted file mode 100644 index f9b52487f..000000000 --- a/internal/varlogcli/stress.go +++ /dev/null @@ -1,67 +0,0 @@ -package varlogcli - -import ( - "context" - "fmt" - "sync" - "time" - - "google.golang.org/grpc" - - "github.com/kakao/varlog/pkg/types" - "github.com/kakao/varlog/proto/snpb" -) - -func AppendStress(snAddr string, tpid types.TopicID, lsid types.LogStreamID, msgSize, batchSize, concurrency, appendCount int) error { - cc, err := grpc.Dial(snAddr, - grpc.WithReadBufferSize(128*1024), - grpc.WithWriteBufferSize(128*1024), - ) - if err != nil { - return err - } - - batch := make([][]byte, batchSize) - for i := 0; i < batchSize; i++ { - msg := make([]byte, msgSize) - for j := 0; j < msgSize; j++ { - msg[j] = '.' - } - batch[i] = msg - } - payloadBytes := int64(msgSize * batchSize) - - fmt.Println("okCnt\terrCnt\tduration\ttotalBytes") - var wg sync.WaitGroup - wg.Add(concurrency) - for i := 0; i < concurrency; i++ { - go func() { - startTime := time.Now() - okCnt, errCnt := int64(0), int64(0) - - defer func() { - duration := time.Since(startTime).Seconds() - totalBytes := payloadBytes * okCnt - fmt.Printf("%d\t%d\t%v\t%d\n", - okCnt, errCnt, duration, totalBytes) - wg.Done() - }() - - client := snpb.NewLogIOClient(cc) - for j := 0; j < appendCount; j++ { - _, err := client.Append(context.TODO(), &snpb.AppendRequest{ - TopicID: tpid, - LogStreamID: lsid, - Payload: batch, - }) - if err != nil { - errCnt++ - continue - } - okCnt++ - } - }() - } - wg.Wait() - return cc.Close() -} diff --git a/proto/snpb/generate.go b/proto/snpb/generate.go index d436e361e..1f891a9ea 100644 --- a/proto/snpb/generate.go +++ b/proto/snpb/generate.go @@ -1,3 +1,3 @@ package snpb -//go:generate mockgen -build_flags -mod=vendor -package mock -destination mock/snpb_mock.go . ReplicatorClient,ReplicatorServer,Replicator_ReplicateClient,LogIOClient,LogIOServer,LogIO_SubscribeClient,LogIO_SubscribeServer,LogStreamReporterClient,LogStreamReporterServer,ManagementClient,ManagementServer +//go:generate mockgen -build_flags -mod=vendor -package mock -destination mock/snpb_mock.go . ReplicatorClient,ReplicatorServer,Replicator_ReplicateClient,LogIOClient,LogIOServer,LogIO_AppendClient,LogIO_SubscribeClient,LogIO_SubscribeServer,LogStreamReporterClient,LogStreamReporterServer,ManagementClient,ManagementServer diff --git a/proto/snpb/log_io.pb.go b/proto/snpb/log_io.pb.go index f8d4382de..fad557e11 100644 --- a/proto/snpb/log_io.pb.go +++ b/proto/snpb/log_io.pb.go @@ -824,64 +824,64 @@ func init() { func init() { proto.RegisterFile("proto/snpb/log_io.proto", fileDescriptor_7692726f23e518ee) } var fileDescriptor_7692726f23e518ee = []byte{ - // 903 bytes of a gzipped FileDescriptorProto + // 907 bytes of a gzipped FileDescriptorProto 0x1f, 0x8b, 0x08, 0x00, 0x00, 0x00, 0x00, 0x00, 0x02, 0xff, 0xdc, 0x57, 0x41, 0x8f, 0xdb, 0x44, - 0x14, 0x8e, 0x13, 0x6f, 0xb3, 0x79, 0xd9, 0x56, 0x65, 0x96, 0xd2, 0xc4, 0x55, 0xe3, 0xc8, 0x42, - 0x68, 0x91, 0x58, 0x1b, 0x05, 0xa1, 0x82, 0x54, 0x24, 0x08, 0x9b, 0xa2, 0x15, 0xe9, 0x82, 0x9c, - 0xa8, 0x07, 0x24, 0x58, 0xd9, 0xf1, 0x60, 0xac, 0x9d, 0x78, 0x8c, 0xed, 0x20, 0x45, 0xfc, 0x00, - 0xae, 0xfd, 0x09, 0xfc, 0x08, 0x0e, 0x5c, 0xb8, 0xf7, 0xd8, 0x0b, 0x12, 0x07, 0x94, 0x43, 0xc2, - 0x7f, 0x40, 0x54, 0x1c, 0xd0, 0x8c, 0xc7, 0x8e, 0xbd, 0x4d, 0xb4, 0x1b, 0xd1, 0x1c, 0x76, 0x6f, - 0x9e, 0x99, 0xf7, 0xbe, 0x79, 0xef, 0x7b, 0x9f, 0xde, 0x3c, 0xc3, 0xdd, 0x20, 0xa4, 0x31, 0x35, - 0x22, 0x3f, 0xb0, 0x0d, 0x42, 0xdd, 0x53, 0x8f, 0xea, 0x7c, 0x07, 0xd5, 0x7f, 0xb0, 0x42, 0x42, - 0x5d, 0x9d, 0x9d, 0x28, 0x87, 0xae, 0x17, 0x7f, 0x37, 0xb1, 0xf5, 0x11, 0x1d, 0x1b, 0x2e, 0x75, - 0xa9, 0xc1, 0x6d, 0xec, 0xc9, 0xb7, 0x7c, 0x95, 0x40, 0xb0, 0xaf, 0xc4, 0x57, 0xb9, 0xe7, 0x52, - 0xea, 0x12, 0xbc, 0xb4, 0xc2, 0xe3, 0x20, 0x9e, 0x8a, 0xc3, 0xbb, 0x09, 0x70, 0x60, 0x1b, 0x63, - 0x1c, 0x5b, 0x8e, 0x15, 0x5b, 0xe2, 0x60, 0x9f, 0x07, 0x51, 0xdc, 0xd4, 0xfe, 0x92, 0xe0, 0xe6, - 0x27, 0x41, 0x80, 0x7d, 0xc7, 0xc4, 0xdf, 0x4f, 0x70, 0x14, 0xa3, 0x01, 0xec, 0xc6, 0x34, 0xf0, - 0x46, 0xa7, 0x9e, 0xd3, 0x90, 0xda, 0xd2, 0xc1, 0x4e, 0xf7, 0x83, 0xf9, 0x4c, 0xad, 0x0e, 0xd9, - 0xde, 0xf1, 0xd1, 0x8b, 0x99, 0xfa, 0x76, 0x2e, 0xd8, 0x33, 0xeb, 0xcc, 0xa2, 0x46, 0x72, 0xa3, - 0x11, 0x9c, 0xb9, 0x46, 0x3c, 0x0d, 0x70, 0xa4, 0x0b, 0x63, 0xb3, 0xca, 0x91, 0x8e, 0x1d, 0xe4, - 0xc0, 0x4d, 0x96, 0x7d, 0x14, 0x87, 0xd8, 0x1a, 0x33, 0xe4, 0x32, 0x47, 0xfe, 0x78, 0x3e, 0x53, - 0xeb, 0x7d, 0xea, 0x0e, 0xf8, 0x3e, 0x47, 0x3f, 0xbc, 0x18, 0x3d, 0xe7, 0x60, 0xd6, 0x49, 0xb6, - 0x70, 0x50, 0x03, 0xaa, 0x81, 0x35, 0x25, 0xd4, 0x72, 0x1a, 0x95, 0x76, 0xe5, 0x60, 0xcf, 0x4c, - 0x97, 0xda, 0xd7, 0xb0, 0x97, 0x66, 0x19, 0x4d, 0x48, 0x8c, 0x1e, 0x80, 0xcc, 0x88, 0xe0, 0x09, - 0xd6, 0x3b, 0xf7, 0x75, 0x51, 0x8c, 0x94, 0x3a, 0x76, 0x45, 0xcf, 0x8f, 0xc3, 0xe9, 0x63, 0x1c, - 0x5b, 0x5d, 0xf9, 0xd9, 0x4c, 0x2d, 0x99, 0xdc, 0x01, 0xbd, 0x0e, 0x3b, 0x38, 0x0c, 0x69, 0xc8, - 0x13, 0xa8, 0x99, 0xc9, 0x42, 0xfb, 0x1c, 0x6e, 0x65, 0xf0, 0x01, 0xf5, 0x23, 0x8c, 0x3e, 0x84, - 0x6a, 0xc8, 0xaf, 0x8a, 0x1a, 0x52, 0xbb, 0x72, 0x50, 0xef, 0x34, 0xf5, 0x5c, 0xc1, 0xf5, 0x7c, - 0x30, 0x02, 0x3f, 0xb5, 0xd7, 0x9e, 0x96, 0xa1, 0x6e, 0x62, 0x2b, 0x2b, 0xc8, 0x23, 0x90, 0x5d, - 0x12, 0xf9, 0x3c, 0x56, 0xb9, 0xdb, 0x99, 0xcf, 0x54, 0xf9, 0xb3, 0xfe, 0xe0, 0xe4, 0xc5, 0x4c, - 0x7d, 0xeb, 0x62, 0xae, 0x98, 0xa5, 0xc9, 0xfd, 0x0b, 0x85, 0x2d, 0x6f, 0xad, 0xb0, 0x95, 0x2d, - 0x14, 0x56, 0xfb, 0x55, 0x82, 0xbd, 0x84, 0x12, 0x41, 0xef, 0xab, 0xe2, 0xe4, 0x11, 0xc8, 0x84, - 0xe1, 0x94, 0x97, 0x38, 0xfd, 0x4b, 0xe3, 0xf4, 0x39, 0x0e, 0xf3, 0x2f, 0x2a, 0x4f, 0xca, 0x2b, - 0xef, 0xef, 0x32, 0xdc, 0x1e, 0x4c, 0xec, 0x68, 0x14, 0x7a, 0x36, 0x4e, 0x4b, 0xfa, 0x04, 0x80, - 0x5d, 0x7f, 0x6a, 0x63, 0xd7, 0x4b, 0x93, 0x78, 0x30, 0x9f, 0xa9, 0x35, 0x16, 0x5a, 0x97, 0x6d, - 0x6e, 0x90, 0x49, 0x8d, 0x41, 0x71, 0x27, 0xf4, 0x25, 0xec, 0x72, 0x5c, 0xec, 0x3b, 0x22, 0xa5, - 0xf7, 0x59, 0x89, 0x99, 0x59, 0xcf, 0x77, 0x36, 0xc0, 0xac, 0x32, 0x98, 0x9e, 0xef, 0x14, 0x44, - 0x53, 0xd9, 0x9a, 0x68, 0xe4, 0x6d, 0x88, 0xe6, 0x37, 0x09, 0x5e, 0xcb, 0x31, 0x7f, 0xe5, 0x94, - 0xf3, 0x4f, 0x19, 0x50, 0x16, 0xff, 0x90, 0x5e, 0x83, 0xfe, 0xfc, 0x04, 0x80, 0x2c, 0x65, 0x5f, - 0x59, 0xca, 0xbe, 0xbf, 0x99, 0xec, 0x39, 0x7d, 0x35, 0x92, 0x97, 0x3d, 0x49, 0x65, 0x2f, 0x2f, - 0x65, 0xdf, 0xdf, 0x44, 0xf6, 0x1c, 0xb3, 0x4a, 0x12, 0xd9, 0x6b, 0x03, 0xd8, 0x2f, 0x50, 0x2f, - 0xc4, 0xf3, 0x10, 0x6a, 0x8c, 0x26, 0xcc, 0x9e, 0x06, 0xf1, 0x76, 0x34, 0xd7, 0xbe, 0x1d, 0xa2, - 0xaf, 0xef, 0x12, 0xb1, 0xd6, 0x7e, 0x91, 0xe0, 0xce, 0x30, 0xf4, 0xc6, 0x47, 0x38, 0x08, 0xf1, - 0xc8, 0x8a, 0xf1, 0x76, 0xdf, 0xdc, 0x54, 0xe9, 0xe5, 0xff, 0xa7, 0x74, 0xed, 0x77, 0x09, 0x1a, - 0x59, 0x49, 0x1f, 0x8b, 0xf1, 0xe1, 0xea, 0xab, 0x51, 0xfb, 0x11, 0x9a, 0x2b, 0xd2, 0x12, 0x95, - 0xfe, 0x06, 0xee, 0xe4, 0x42, 0x70, 0x30, 0x93, 0x42, 0x10, 0xd3, 0x50, 0x54, 0xfd, 0xcd, 0x55, - 0x55, 0x4f, 0xa0, 0x8e, 0x32, 0x5b, 0x21, 0x80, 0x7d, 0xf2, 0xf2, 0x91, 0xf6, 0xa7, 0x04, 0x6a, - 0xe6, 0x62, 0xe2, 0x80, 0x78, 0x23, 0xeb, 0x1a, 0x71, 0xfb, 0x93, 0x04, 0xed, 0xf5, 0xe9, 0x09, - 0x8e, 0x47, 0x80, 0x72, 0xa1, 0x84, 0x89, 0x95, 0x20, 0xd8, 0x28, 0x8c, 0x4b, 0xeb, 0xa0, 0x5e, - 0xe2, 0xfa, 0x36, 0x39, 0x67, 0xd9, 0xf9, 0xb7, 0x02, 0x3b, 0x7d, 0xea, 0x1e, 0x7f, 0x81, 0x3e, - 0x85, 0x1b, 0xc9, 0xd8, 0x85, 0x94, 0x95, 0xb3, 0x18, 0x27, 0x5d, 0xb9, 0xb7, 0x7a, 0x4e, 0xe3, - 0x11, 0x6b, 0x25, 0xf4, 0x11, 0xc8, 0x6c, 0x10, 0x41, 0x8d, 0x82, 0x59, 0x6e, 0x5c, 0x53, 0x9a, - 0x2b, 0x4e, 0x32, 0xf7, 0x13, 0xa8, 0x65, 0x7d, 0x05, 0xdd, 0x2f, 0x58, 0x9e, 0x1f, 0x12, 0x94, - 0xd6, 0xba, 0xe3, 0x14, 0xed, 0x5d, 0x09, 0x0d, 0xa1, 0x9e, 0xeb, 0x53, 0x48, 0x5d, 0xed, 0x92, - 0x3d, 0x1e, 0x4a, 0x7b, 0xbd, 0x41, 0x0e, 0xf5, 0x04, 0x6e, 0x15, 0xfb, 0x14, 0xd2, 0x0a, 0x7e, - 0x2b, 0x9b, 0x98, 0xf2, 0x86, 0x9e, 0xfc, 0x96, 0xe8, 0xe9, 0x6f, 0x89, 0xde, 0x63, 0xbf, 0x25, - 0x5a, 0x09, 0x4d, 0x73, 0x0d, 0xe4, 0x5c, 0x05, 0xd1, 0x3b, 0x97, 0x2a, 0x74, 0x7a, 0xc7, 0xe1, - 0x25, 0xad, 0xd3, 0x64, 0xba, 0x0f, 0x9f, 0xcd, 0x5b, 0xd2, 0xf3, 0x79, 0x4b, 0x7a, 0xba, 0x68, - 0x95, 0x7e, 0x5e, 0xb4, 0xa4, 0xe7, 0x8b, 0x56, 0xe9, 0x8f, 0x45, 0xab, 0xf4, 0x95, 0xb6, 0x56, - 0xde, 0xd9, 0x1f, 0x9b, 0x7d, 0x83, 0x7f, 0xbf, 0xf7, 0x5f, 0x00, 0x00, 0x00, 0xff, 0xff, 0xb7, - 0xee, 0x16, 0xb1, 0xc6, 0x0d, 0x00, 0x00, + 0x14, 0xce, 0x24, 0xde, 0x66, 0xf3, 0xb2, 0xad, 0xca, 0x2c, 0xa5, 0x59, 0x57, 0x8d, 0x23, 0x0b, + 0xa1, 0x45, 0x62, 0xed, 0x6a, 0x11, 0x2a, 0x48, 0x45, 0x82, 0x68, 0xb7, 0xd5, 0x8a, 0x74, 0x41, + 0x4e, 0xd4, 0x03, 0x12, 0xac, 0xec, 0x78, 0x30, 0xd6, 0x4e, 0x3c, 0xc6, 0x76, 0x90, 0x22, 0x2e, + 0xdc, 0xb8, 0xf6, 0x27, 0xf0, 0x23, 0x38, 0x70, 0xe1, 0xde, 0x63, 0x2f, 0x48, 0x1c, 0x50, 0x0e, + 0x09, 0xff, 0x01, 0xd1, 0x13, 0x9a, 0xf1, 0xd8, 0xb1, 0xb7, 0x89, 0xba, 0x11, 0xe4, 0xd0, 0xbd, + 0x79, 0x66, 0xde, 0xfb, 0xe6, 0xbd, 0xef, 0x7d, 0x7a, 0xf3, 0x0c, 0xb7, 0xc3, 0x88, 0x25, 0xcc, + 0x8c, 0x83, 0xd0, 0x31, 0x29, 0xf3, 0xce, 0x7c, 0x66, 0x88, 0x1d, 0xdc, 0xfc, 0xde, 0x8e, 0x28, + 0xf3, 0x0c, 0x7e, 0xa2, 0x1e, 0x78, 0x7e, 0xf2, 0xed, 0xd8, 0x31, 0x86, 0x6c, 0x64, 0x7a, 0xcc, + 0x63, 0xa6, 0xb0, 0x71, 0xc6, 0xdf, 0x88, 0x55, 0x0a, 0xc1, 0xbf, 0x52, 0x5f, 0xf5, 0x8e, 0xc7, + 0x98, 0x47, 0xc9, 0xc2, 0x8a, 0x8c, 0xc2, 0x64, 0x22, 0x0f, 0x6f, 0xa7, 0xc0, 0xa1, 0x63, 0x8e, + 0x48, 0x62, 0xbb, 0x76, 0x62, 0xcb, 0x83, 0x5d, 0x11, 0x44, 0x79, 0x53, 0xff, 0x0b, 0xc1, 0xf5, + 0x4f, 0xc3, 0x90, 0x04, 0xae, 0x45, 0xbe, 0x1b, 0x93, 0x38, 0xc1, 0x7d, 0xd8, 0x4e, 0x58, 0xe8, + 0x0f, 0xcf, 0x7c, 0xb7, 0x85, 0x3a, 0x68, 0x7f, 0xab, 0xfb, 0xe1, 0x6c, 0xaa, 0xd5, 0x07, 0x7c, + 0xef, 0xe4, 0xe8, 0xc5, 0x54, 0x7b, 0xb7, 0x10, 0xec, 0xb9, 0x7d, 0x6e, 0x33, 0x33, 0xbd, 0xd1, + 0x0c, 0xcf, 0x3d, 0x33, 0x99, 0x84, 0x24, 0x36, 0xa4, 0xb1, 0x55, 0x17, 0x48, 0x27, 0x2e, 0x76, + 0xe1, 0x3a, 0xcf, 0x3e, 0x4e, 0x22, 0x62, 0x8f, 0x38, 0x72, 0x55, 0x20, 0x7f, 0x32, 0x9b, 0x6a, + 0xcd, 0x1e, 0xf3, 0xfa, 0x62, 0x5f, 0xa0, 0x1f, 0xbc, 0x1a, 0xbd, 0xe0, 0x60, 0x35, 0x69, 0xbe, + 0x70, 0x71, 0x0b, 0xea, 0xa1, 0x3d, 0xa1, 0xcc, 0x76, 0x5b, 0xb5, 0x4e, 0x6d, 0x7f, 0xc7, 0xca, + 0x96, 0xfa, 0x57, 0xb0, 0x93, 0x65, 0x19, 0x8f, 0x69, 0x82, 0xef, 0x83, 0xc2, 0x89, 0x10, 0x09, + 0x36, 0x0f, 0xef, 0x1a, 0xb2, 0x18, 0x19, 0x75, 0xfc, 0x8a, 0xe3, 0x20, 0x89, 0x26, 0x8f, 0x49, + 0x62, 0x77, 0x95, 0x67, 0x53, 0xad, 0x62, 0x09, 0x07, 0xfc, 0x26, 0x6c, 0x91, 0x28, 0x62, 0x91, + 0x48, 0xa0, 0x61, 0xa5, 0x0b, 0xfd, 0x33, 0xb8, 0x91, 0xc3, 0x87, 0x2c, 0x88, 0x09, 0xfe, 0x08, + 0xea, 0x91, 0xb8, 0x2a, 0x6e, 0xa1, 0x4e, 0x6d, 0xbf, 0x79, 0xb8, 0x67, 0x14, 0x0a, 0x6e, 0x14, + 0x83, 0x91, 0xf8, 0x99, 0xbd, 0xfe, 0xb4, 0x0a, 0x4d, 0x8b, 0xd8, 0x79, 0x41, 0x1e, 0x82, 0xe2, + 0xd1, 0x38, 0x10, 0xb1, 0x2a, 0xdd, 0xc3, 0xd9, 0x54, 0x53, 0x1e, 0xf5, 0xfa, 0xa7, 0x2f, 0xa6, + 0xda, 0x3b, 0xaf, 0xe6, 0x8a, 0x5b, 0x5a, 0xc2, 0xbf, 0x54, 0xd8, 0xea, 0xc6, 0x0a, 0x5b, 0xdb, + 0x40, 0x61, 0xf5, 0x5f, 0x11, 0xec, 0xa4, 0x94, 0x48, 0x7a, 0xff, 0x2f, 0x4e, 0x1e, 0x82, 0x42, + 0x39, 0x4e, 0x75, 0x81, 0xd3, 0xbb, 0x34, 0x4e, 0x4f, 0xe0, 0x70, 0xff, 0xb2, 0xf2, 0x50, 0x51, + 0x79, 0x7f, 0x57, 0xe1, 0x66, 0x7f, 0xec, 0xc4, 0xc3, 0xc8, 0x77, 0x48, 0x56, 0xd2, 0x27, 0x00, + 0xfc, 0xfa, 0x33, 0x87, 0x78, 0x7e, 0x96, 0xc4, 0xfd, 0xd9, 0x54, 0x6b, 0xf0, 0xd0, 0xba, 0x7c, + 0x73, 0x8d, 0x4c, 0x1a, 0x1c, 0x4a, 0x38, 0xe1, 0x2f, 0x60, 0x5b, 0xe0, 0x92, 0xc0, 0x95, 0x29, + 0x7d, 0xc0, 0x4b, 0xcc, 0xcd, 0x8e, 0x03, 0x77, 0x0d, 0xcc, 0x3a, 0x87, 0x39, 0x0e, 0xdc, 0x92, + 0x68, 0x6a, 0x1b, 0x13, 0x8d, 0xb2, 0x09, 0xd1, 0xfc, 0x86, 0xe0, 0x8d, 0x02, 0xf3, 0xaf, 0x9d, + 0x72, 0xfe, 0xa9, 0x02, 0xce, 0xe3, 0x1f, 0xb0, 0x2b, 0xd0, 0x9f, 0x9f, 0x00, 0xd0, 0x85, 0xec, + 0x6b, 0x0b, 0xd9, 0xf7, 0xd6, 0x93, 0xbd, 0xa0, 0xaf, 0x41, 0x8b, 0xb2, 0xa7, 0x99, 0xec, 0x95, + 0x85, 0xec, 0x7b, 0xeb, 0xc8, 0x5e, 0x60, 0xd6, 0x69, 0x2a, 0x7b, 0xbd, 0x0f, 0xbb, 0x25, 0xea, + 0xa5, 0x78, 0x1e, 0x40, 0x83, 0xd3, 0x44, 0xf8, 0xd3, 0x20, 0xdf, 0x8e, 0xbd, 0x95, 0x6f, 0x87, + 0xec, 0xeb, 0xdb, 0x54, 0xae, 0xf5, 0x5f, 0x10, 0xdc, 0x1a, 0x44, 0xfe, 0xe8, 0x88, 0x84, 0x11, + 0x19, 0xda, 0x09, 0xd9, 0xec, 0x9b, 0x9b, 0x29, 0xbd, 0xfa, 0xdf, 0x94, 0xae, 0xff, 0x8e, 0xa0, + 0x95, 0x97, 0xf4, 0xb1, 0x1c, 0x1f, 0x5e, 0x7f, 0x35, 0xea, 0x3f, 0xc0, 0xde, 0x92, 0xb4, 0x64, + 0xa5, 0xbf, 0x86, 0x5b, 0x85, 0x10, 0x5c, 0xc2, 0xa5, 0x10, 0x26, 0x2c, 0x92, 0x55, 0x7f, 0x7b, + 0x59, 0xd5, 0x53, 0xa8, 0xa3, 0xdc, 0x56, 0x0a, 0x60, 0x97, 0xbe, 0x7c, 0xa4, 0xff, 0x89, 0x40, + 0xcb, 0x5d, 0x2c, 0x12, 0x52, 0x7f, 0x68, 0x5f, 0x21, 0x6e, 0x7f, 0x42, 0xd0, 0x59, 0x9d, 0x9e, + 0xe4, 0x78, 0x08, 0xb8, 0x10, 0x4a, 0x94, 0x5a, 0x49, 0x82, 0xcd, 0xd2, 0xb8, 0xb4, 0x0a, 0xea, + 0x25, 0xae, 0x6f, 0xd2, 0x0b, 0x96, 0x87, 0x3f, 0x2a, 0xb0, 0xd5, 0x63, 0xde, 0xc9, 0xe7, 0xf8, + 0x11, 0x5c, 0x4b, 0xc7, 0x2e, 0xac, 0x2e, 0x9d, 0xc5, 0x04, 0xe9, 0xea, 0x9d, 0xe5, 0x73, 0x9a, + 0x88, 0x58, 0xaf, 0xec, 0xa3, 0x7b, 0x08, 0x7f, 0x0c, 0x0a, 0x1f, 0x46, 0x70, 0xab, 0x64, 0x5a, + 0x18, 0xd9, 0xd4, 0xbd, 0x25, 0x27, 0x19, 0x04, 0x3e, 0x85, 0x46, 0xde, 0x5b, 0xf0, 0xdd, 0x92, + 0xe5, 0xc5, 0x41, 0x41, 0x6d, 0xaf, 0x3a, 0xce, 0xd0, 0xee, 0x21, 0x3c, 0x80, 0x66, 0xa1, 0x57, + 0x61, 0x6d, 0xb9, 0x4b, 0xfe, 0x80, 0xa8, 0x9d, 0xd5, 0x06, 0x05, 0xd4, 0x53, 0xb8, 0x51, 0xee, + 0x55, 0x58, 0x2f, 0xf9, 0x2d, 0x6d, 0x64, 0xea, 0x5b, 0x46, 0xfa, 0x6b, 0x62, 0x64, 0xbf, 0x26, + 0xc6, 0x31, 0xff, 0x35, 0xd1, 0x2b, 0x78, 0x52, 0x68, 0x22, 0x17, 0xaa, 0x88, 0xdf, 0xbb, 0x54, + 0xb1, 0xb3, 0x3b, 0x0e, 0x2e, 0x69, 0x9d, 0x25, 0xd3, 0x7d, 0xf0, 0x6c, 0xd6, 0x46, 0xcf, 0x67, + 0x6d, 0xf4, 0x74, 0xde, 0xae, 0xfc, 0x3c, 0x6f, 0xa3, 0xe7, 0xf3, 0x76, 0xe5, 0x8f, 0x79, 0xbb, + 0xf2, 0xa5, 0xbe, 0x52, 0xe2, 0xf9, 0x5f, 0x9b, 0x73, 0x4d, 0x7c, 0xbf, 0xff, 0x6f, 0x00, 0x00, + 0x00, 0xff, 0xff, 0x09, 0x66, 0x9b, 0x7b, 0xca, 0x0d, 0x00, 0x00, } // Reference imports to suppress errors if they are not otherwise used. @@ -896,9 +896,9 @@ const _ = grpc.SupportPackageIsVersion4 // // For semantics around ctx use and closing/ending streaming RPCs, please refer to https://godoc.org/google.golang.org/grpc#ClientConn.NewStream. type LogIOClient interface { - // Append stores a list of log entries to the end of the log stream specified - // by AppendRequest. The log entries are appended partially; that is, some of - // the log entries could not be stored due to failures. + // Append stores a list of log entries to the end of the log stream + // specified by AppendRequest. The log entries are appended partially; that + // is, some of the log entries could not be stored due to failures. // // It returns the following gRPC errors: // - InvalidArgument: AppendRequest has invalid fields; for instance, TopicID @@ -915,7 +915,7 @@ type LogIOClient interface { // - DeadlineExceeded: The client's timeout has expired. // // FIXME: Partial failures are not specified by the gRPC error codes. - Append(ctx context.Context, in *AppendRequest, opts ...grpc.CallOption) (*AppendResponse, error) + Append(ctx context.Context, opts ...grpc.CallOption) (LogIO_AppendClient, error) // Read reads a log entry from the log stream specified by ReadRequest. // Deprecated: Use Subscribe or SubscribeTo. Read(ctx context.Context, in *ReadRequest, opts ...grpc.CallOption) (*ReadResponse, error) @@ -947,13 +947,35 @@ func NewLogIOClient(cc *grpc.ClientConn) LogIOClient { return &logIOClient{cc} } -func (c *logIOClient) Append(ctx context.Context, in *AppendRequest, opts ...grpc.CallOption) (*AppendResponse, error) { - out := new(AppendResponse) - err := c.cc.Invoke(ctx, "/varlog.snpb.LogIO/Append", in, out, opts...) +func (c *logIOClient) Append(ctx context.Context, opts ...grpc.CallOption) (LogIO_AppendClient, error) { + stream, err := c.cc.NewStream(ctx, &_LogIO_serviceDesc.Streams[0], "/varlog.snpb.LogIO/Append", opts...) if err != nil { return nil, err } - return out, nil + x := &logIOAppendClient{stream} + return x, nil +} + +type LogIO_AppendClient interface { + Send(*AppendRequest) error + Recv() (*AppendResponse, error) + grpc.ClientStream +} + +type logIOAppendClient struct { + grpc.ClientStream +} + +func (x *logIOAppendClient) Send(m *AppendRequest) error { + return x.ClientStream.SendMsg(m) +} + +func (x *logIOAppendClient) Recv() (*AppendResponse, error) { + m := new(AppendResponse) + if err := x.ClientStream.RecvMsg(m); err != nil { + return nil, err + } + return m, nil } func (c *logIOClient) Read(ctx context.Context, in *ReadRequest, opts ...grpc.CallOption) (*ReadResponse, error) { @@ -966,7 +988,7 @@ func (c *logIOClient) Read(ctx context.Context, in *ReadRequest, opts ...grpc.Ca } func (c *logIOClient) Subscribe(ctx context.Context, in *SubscribeRequest, opts ...grpc.CallOption) (LogIO_SubscribeClient, error) { - stream, err := c.cc.NewStream(ctx, &_LogIO_serviceDesc.Streams[0], "/varlog.snpb.LogIO/Subscribe", opts...) + stream, err := c.cc.NewStream(ctx, &_LogIO_serviceDesc.Streams[1], "/varlog.snpb.LogIO/Subscribe", opts...) if err != nil { return nil, err } @@ -998,7 +1020,7 @@ func (x *logIOSubscribeClient) Recv() (*SubscribeResponse, error) { } func (c *logIOClient) SubscribeTo(ctx context.Context, in *SubscribeToRequest, opts ...grpc.CallOption) (LogIO_SubscribeToClient, error) { - stream, err := c.cc.NewStream(ctx, &_LogIO_serviceDesc.Streams[1], "/varlog.snpb.LogIO/SubscribeTo", opts...) + stream, err := c.cc.NewStream(ctx, &_LogIO_serviceDesc.Streams[2], "/varlog.snpb.LogIO/SubscribeTo", opts...) if err != nil { return nil, err } @@ -1049,9 +1071,9 @@ func (c *logIOClient) LogStreamReplicaMetadata(ctx context.Context, in *LogStrea // LogIOServer is the server API for LogIO service. type LogIOServer interface { - // Append stores a list of log entries to the end of the log stream specified - // by AppendRequest. The log entries are appended partially; that is, some of - // the log entries could not be stored due to failures. + // Append stores a list of log entries to the end of the log stream + // specified by AppendRequest. The log entries are appended partially; that + // is, some of the log entries could not be stored due to failures. // // It returns the following gRPC errors: // - InvalidArgument: AppendRequest has invalid fields; for instance, TopicID @@ -1068,7 +1090,7 @@ type LogIOServer interface { // - DeadlineExceeded: The client's timeout has expired. // // FIXME: Partial failures are not specified by the gRPC error codes. - Append(context.Context, *AppendRequest) (*AppendResponse, error) + Append(LogIO_AppendServer) error // Read reads a log entry from the log stream specified by ReadRequest. // Deprecated: Use Subscribe or SubscribeTo. Read(context.Context, *ReadRequest) (*ReadResponse, error) @@ -1096,8 +1118,8 @@ type LogIOServer interface { type UnimplementedLogIOServer struct { } -func (*UnimplementedLogIOServer) Append(ctx context.Context, req *AppendRequest) (*AppendResponse, error) { - return nil, status.Errorf(codes.Unimplemented, "method Append not implemented") +func (*UnimplementedLogIOServer) Append(srv LogIO_AppendServer) error { + return status.Errorf(codes.Unimplemented, "method Append not implemented") } func (*UnimplementedLogIOServer) Read(ctx context.Context, req *ReadRequest) (*ReadResponse, error) { return nil, status.Errorf(codes.Unimplemented, "method Read not implemented") @@ -1119,22 +1141,30 @@ func RegisterLogIOServer(s *grpc.Server, srv LogIOServer) { s.RegisterService(&_LogIO_serviceDesc, srv) } -func _LogIO_Append_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) { - in := new(AppendRequest) - if err := dec(in); err != nil { +func _LogIO_Append_Handler(srv interface{}, stream grpc.ServerStream) error { + return srv.(LogIOServer).Append(&logIOAppendServer{stream}) +} + +type LogIO_AppendServer interface { + Send(*AppendResponse) error + Recv() (*AppendRequest, error) + grpc.ServerStream +} + +type logIOAppendServer struct { + grpc.ServerStream +} + +func (x *logIOAppendServer) Send(m *AppendResponse) error { + return x.ServerStream.SendMsg(m) +} + +func (x *logIOAppendServer) Recv() (*AppendRequest, error) { + m := new(AppendRequest) + if err := x.ServerStream.RecvMsg(m); err != nil { return nil, err } - if interceptor == nil { - return srv.(LogIOServer).Append(ctx, in) - } - info := &grpc.UnaryServerInfo{ - Server: srv, - FullMethod: "/varlog.snpb.LogIO/Append", - } - handler := func(ctx context.Context, req interface{}) (interface{}, error) { - return srv.(LogIOServer).Append(ctx, req.(*AppendRequest)) - } - return interceptor(ctx, in, info, handler) + return m, nil } func _LogIO_Read_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) { @@ -1237,10 +1267,6 @@ var _LogIO_serviceDesc = grpc.ServiceDesc{ ServiceName: "varlog.snpb.LogIO", HandlerType: (*LogIOServer)(nil), Methods: []grpc.MethodDesc{ - { - MethodName: "Append", - Handler: _LogIO_Append_Handler, - }, { MethodName: "Read", Handler: _LogIO_Read_Handler, @@ -1255,6 +1281,12 @@ var _LogIO_serviceDesc = grpc.ServiceDesc{ }, }, Streams: []grpc.StreamDesc{ + { + StreamName: "Append", + Handler: _LogIO_Append_Handler, + ServerStreams: true, + ClientStreams: true, + }, { StreamName: "Subscribe", Handler: _LogIO_Subscribe_Handler, diff --git a/proto/snpb/log_io.proto b/proto/snpb/log_io.proto index e0efad5c1..0927643b6 100644 --- a/proto/snpb/log_io.proto +++ b/proto/snpb/log_io.proto @@ -175,9 +175,9 @@ message LogStreamReplicaMetadataResponse { } service LogIO { - // Append stores a list of log entries to the end of the log stream specified - // by AppendRequest. The log entries are appended partially; that is, some of - // the log entries could not be stored due to failures. + // Append stores a list of log entries to the end of the log stream + // specified by AppendRequest. The log entries are appended partially; that + // is, some of the log entries could not be stored due to failures. // // It returns the following gRPC errors: // - InvalidArgument: AppendRequest has invalid fields; for instance, TopicID @@ -194,7 +194,8 @@ service LogIO { // - DeadlineExceeded: The client's timeout has expired. // // FIXME: Partial failures are not specified by the gRPC error codes. - rpc Append(AppendRequest) returns (AppendResponse) {} + rpc Append(stream AppendRequest) returns (stream AppendResponse) {} + // Read reads a log entry from the log stream specified by ReadRequest. // Deprecated: Use Subscribe or SubscribeTo. rpc Read(ReadRequest) returns (ReadResponse) {} diff --git a/proto/snpb/mock/snpb_mock.go b/proto/snpb/mock/snpb_mock.go index 21a3d3763..962f536d9 100644 --- a/proto/snpb/mock/snpb_mock.go +++ b/proto/snpb/mock/snpb_mock.go @@ -1,5 +1,5 @@ // Code generated by MockGen. DO NOT EDIT. -// Source: github.com/kakao/varlog/proto/snpb (interfaces: ReplicatorClient,ReplicatorServer,Replicator_ReplicateClient,LogIOClient,LogIOServer,LogIO_SubscribeClient,LogIO_SubscribeServer,LogStreamReporterClient,LogStreamReporterServer,ManagementClient,ManagementServer) +// Source: github.com/kakao/varlog/proto/snpb (interfaces: ReplicatorClient,ReplicatorServer,Replicator_ReplicateClient,LogIOClient,LogIOServer,LogIO_AppendClient,LogIO_SubscribeClient,LogIO_SubscribeServer,LogStreamReporterClient,LogStreamReporterServer,ManagementClient,ManagementServer) // Package mock is a generated GoMock package. package mock @@ -361,22 +361,22 @@ func (m *MockLogIOClient) EXPECT() *MockLogIOClientMockRecorder { } // Append mocks base method. -func (m *MockLogIOClient) Append(arg0 context.Context, arg1 *snpb.AppendRequest, arg2 ...grpc.CallOption) (*snpb.AppendResponse, error) { +func (m *MockLogIOClient) Append(arg0 context.Context, arg1 ...grpc.CallOption) (snpb.LogIO_AppendClient, error) { m.ctrl.T.Helper() - varargs := []interface{}{arg0, arg1} - for _, a := range arg2 { + varargs := []interface{}{arg0} + for _, a := range arg1 { varargs = append(varargs, a) } ret := m.ctrl.Call(m, "Append", varargs...) - ret0, _ := ret[0].(*snpb.AppendResponse) + ret0, _ := ret[0].(snpb.LogIO_AppendClient) ret1, _ := ret[1].(error) return ret0, ret1 } // Append indicates an expected call of Append. -func (mr *MockLogIOClientMockRecorder) Append(arg0, arg1 interface{}, arg2 ...interface{}) *gomock.Call { +func (mr *MockLogIOClientMockRecorder) Append(arg0 interface{}, arg1 ...interface{}) *gomock.Call { mr.mock.ctrl.T.Helper() - varargs := append([]interface{}{arg0, arg1}, arg2...) + varargs := append([]interface{}{arg0}, arg1...) return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Append", reflect.TypeOf((*MockLogIOClient)(nil).Append), varargs...) } @@ -504,18 +504,17 @@ func (m *MockLogIOServer) EXPECT() *MockLogIOServerMockRecorder { } // Append mocks base method. -func (m *MockLogIOServer) Append(arg0 context.Context, arg1 *snpb.AppendRequest) (*snpb.AppendResponse, error) { +func (m *MockLogIOServer) Append(arg0 snpb.LogIO_AppendServer) error { m.ctrl.T.Helper() - ret := m.ctrl.Call(m, "Append", arg0, arg1) - ret0, _ := ret[0].(*snpb.AppendResponse) - ret1, _ := ret[1].(error) - return ret0, ret1 + ret := m.ctrl.Call(m, "Append", arg0) + ret0, _ := ret[0].(error) + return ret0 } // Append indicates an expected call of Append. -func (mr *MockLogIOServerMockRecorder) Append(arg0, arg1 interface{}) *gomock.Call { +func (mr *MockLogIOServerMockRecorder) Append(arg0 interface{}) *gomock.Call { mr.mock.ctrl.T.Helper() - return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Append", reflect.TypeOf((*MockLogIOServer)(nil).Append), arg0, arg1) + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Append", reflect.TypeOf((*MockLogIOServer)(nil).Append), arg0) } // LogStreamReplicaMetadata mocks base method. @@ -591,6 +590,143 @@ func (mr *MockLogIOServerMockRecorder) TrimDeprecated(arg0, arg1 interface{}) *g return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "TrimDeprecated", reflect.TypeOf((*MockLogIOServer)(nil).TrimDeprecated), arg0, arg1) } +// MockLogIO_AppendClient is a mock of LogIO_AppendClient interface. +type MockLogIO_AppendClient struct { + ctrl *gomock.Controller + recorder *MockLogIO_AppendClientMockRecorder +} + +// MockLogIO_AppendClientMockRecorder is the mock recorder for MockLogIO_AppendClient. +type MockLogIO_AppendClientMockRecorder struct { + mock *MockLogIO_AppendClient +} + +// NewMockLogIO_AppendClient creates a new mock instance. +func NewMockLogIO_AppendClient(ctrl *gomock.Controller) *MockLogIO_AppendClient { + mock := &MockLogIO_AppendClient{ctrl: ctrl} + mock.recorder = &MockLogIO_AppendClientMockRecorder{mock} + return mock +} + +// EXPECT returns an object that allows the caller to indicate expected use. +func (m *MockLogIO_AppendClient) EXPECT() *MockLogIO_AppendClientMockRecorder { + return m.recorder +} + +// CloseSend mocks base method. +func (m *MockLogIO_AppendClient) CloseSend() error { + m.ctrl.T.Helper() + ret := m.ctrl.Call(m, "CloseSend") + ret0, _ := ret[0].(error) + return ret0 +} + +// CloseSend indicates an expected call of CloseSend. +func (mr *MockLogIO_AppendClientMockRecorder) CloseSend() *gomock.Call { + mr.mock.ctrl.T.Helper() + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "CloseSend", reflect.TypeOf((*MockLogIO_AppendClient)(nil).CloseSend)) +} + +// Context mocks base method. +func (m *MockLogIO_AppendClient) Context() context.Context { + m.ctrl.T.Helper() + ret := m.ctrl.Call(m, "Context") + ret0, _ := ret[0].(context.Context) + return ret0 +} + +// Context indicates an expected call of Context. +func (mr *MockLogIO_AppendClientMockRecorder) Context() *gomock.Call { + mr.mock.ctrl.T.Helper() + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Context", reflect.TypeOf((*MockLogIO_AppendClient)(nil).Context)) +} + +// Header mocks base method. +func (m *MockLogIO_AppendClient) Header() (metadata.MD, error) { + m.ctrl.T.Helper() + ret := m.ctrl.Call(m, "Header") + ret0, _ := ret[0].(metadata.MD) + ret1, _ := ret[1].(error) + return ret0, ret1 +} + +// Header indicates an expected call of Header. +func (mr *MockLogIO_AppendClientMockRecorder) Header() *gomock.Call { + mr.mock.ctrl.T.Helper() + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Header", reflect.TypeOf((*MockLogIO_AppendClient)(nil).Header)) +} + +// Recv mocks base method. +func (m *MockLogIO_AppendClient) Recv() (*snpb.AppendResponse, error) { + m.ctrl.T.Helper() + ret := m.ctrl.Call(m, "Recv") + ret0, _ := ret[0].(*snpb.AppendResponse) + ret1, _ := ret[1].(error) + return ret0, ret1 +} + +// Recv indicates an expected call of Recv. +func (mr *MockLogIO_AppendClientMockRecorder) Recv() *gomock.Call { + mr.mock.ctrl.T.Helper() + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Recv", reflect.TypeOf((*MockLogIO_AppendClient)(nil).Recv)) +} + +// RecvMsg mocks base method. +func (m *MockLogIO_AppendClient) RecvMsg(arg0 interface{}) error { + m.ctrl.T.Helper() + ret := m.ctrl.Call(m, "RecvMsg", arg0) + ret0, _ := ret[0].(error) + return ret0 +} + +// RecvMsg indicates an expected call of RecvMsg. +func (mr *MockLogIO_AppendClientMockRecorder) RecvMsg(arg0 interface{}) *gomock.Call { + mr.mock.ctrl.T.Helper() + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "RecvMsg", reflect.TypeOf((*MockLogIO_AppendClient)(nil).RecvMsg), arg0) +} + +// Send mocks base method. +func (m *MockLogIO_AppendClient) Send(arg0 *snpb.AppendRequest) error { + m.ctrl.T.Helper() + ret := m.ctrl.Call(m, "Send", arg0) + ret0, _ := ret[0].(error) + return ret0 +} + +// Send indicates an expected call of Send. +func (mr *MockLogIO_AppendClientMockRecorder) Send(arg0 interface{}) *gomock.Call { + mr.mock.ctrl.T.Helper() + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Send", reflect.TypeOf((*MockLogIO_AppendClient)(nil).Send), arg0) +} + +// SendMsg mocks base method. +func (m *MockLogIO_AppendClient) SendMsg(arg0 interface{}) error { + m.ctrl.T.Helper() + ret := m.ctrl.Call(m, "SendMsg", arg0) + ret0, _ := ret[0].(error) + return ret0 +} + +// SendMsg indicates an expected call of SendMsg. +func (mr *MockLogIO_AppendClientMockRecorder) SendMsg(arg0 interface{}) *gomock.Call { + mr.mock.ctrl.T.Helper() + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "SendMsg", reflect.TypeOf((*MockLogIO_AppendClient)(nil).SendMsg), arg0) +} + +// Trailer mocks base method. +func (m *MockLogIO_AppendClient) Trailer() metadata.MD { + m.ctrl.T.Helper() + ret := m.ctrl.Call(m, "Trailer") + ret0, _ := ret[0].(metadata.MD) + return ret0 +} + +// Trailer indicates an expected call of Trailer. +func (mr *MockLogIO_AppendClientMockRecorder) Trailer() *gomock.Call { + mr.mock.ctrl.T.Helper() + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Trailer", reflect.TypeOf((*MockLogIO_AppendClient)(nil).Trailer)) +} + // MockLogIO_SubscribeClient is a mock of LogIO_SubscribeClient interface. type MockLogIO_SubscribeClient struct { ctrl *gomock.Controller