fix review comments

Signed-off-by: jyjiangkai <jyjiangkai@163.com>

Author: hwjiangkai

client/internal/vanus/store/block_store.go

@@ -28,16 +28,17 @@ import (
 	"google.golang.org/grpc"
 
 	// first-party libraries
-  "github.com/linkall-labs/vanus/client/pkg/codec"
 	"github.com/linkall-labs/vanus/observability/log"
 	"github.com/linkall-labs/vanus/observability/tracing"
 	"github.com/linkall-labs/vanus/pkg/errors"
 	cepb "github.com/linkall-labs/vanus/proto/pkg/cloudevents"
 	errpb "github.com/linkall-labs/vanus/proto/pkg/errors"
 	segpb "github.com/linkall-labs/vanus/proto/pkg/segment"
 
+	// this project
 	"github.com/linkall-labs/vanus/client/internal/vanus/net/rpc"
 	"github.com/linkall-labs/vanus/client/internal/vanus/net/rpc/bare"
+	"github.com/linkall-labs/vanus/client/pkg/codec"
 	"github.com/linkall-labs/vanus/client/pkg/primitive"
 )
 
@@ -80,19 +81,31 @@ var (
 type appendStreamCache struct {
 	opaqueID  uint64
 	stream    segpb.SegmentServer_AppendToBlockStreamClient
+	reqs      chan *segpb.AppendToBlockStreamRequest
 	callbacks sync.Map
 	state     streamState
 	once      sync.Once
+	mu        sync.RWMutex
 }
 
 func (a *appendStreamCache) isRunning() bool {
+	a.mu.RLock()
+	defer a.mu.RUnlock()
 	return a.state == stateRunning
 }
 
 func (a *appendStreamCache) isClosed() bool {
+	a.mu.RLock()
+	defer a.mu.RUnlock()
 	return a.state == stateClosed
 }
 
+func (a *appendStreamCache) setState(state streamState) {
+	a.mu.Lock()
+	defer a.mu.Unlock()
+	a.state = state
+}
+
 func (a *appendStreamCache) release() {
 	a.releaseStream()
 	a.releaseCallbacks()
@@ -101,6 +114,7 @@ func (a *appendStreamCache) release() {
 func (a *appendStreamCache) releaseStream() {
 	a.once.Do(func() {
 		a.stream.CloseSend()
+		close(a.reqs)
 		a.state = stateClosed
 	})
 }
@@ -119,19 +133,31 @@ func (a *appendStreamCache) releaseCallbacks() {
 type readStreamCache struct {
 	opaqueID  uint64
 	stream    segpb.SegmentServer_ReadFromBlockStreamClient
+	reqs      chan *segpb.ReadFromBlockStreamRequest
 	callbacks sync.Map
 	state     streamState
 	once      sync.Once
+	mu        sync.RWMutex
 }
 
 func (r *readStreamCache) isRunning() bool {
+	r.mu.RLock()
+	defer r.mu.RUnlock()
 	return r.state == stateRunning
 }
 
 func (r *readStreamCache) isClosed() bool {
+	r.mu.RLock()
+	defer r.mu.RUnlock()
 	return r.state == stateClosed
 }
 
+func (r *readStreamCache) setState(state streamState) {
+	r.mu.Lock()
+	defer r.mu.Unlock()
+	r.state = state
+}
+
 func (r *readStreamCache) release() {
 	r.releaseStream()
 	r.releaseCallbacks()
@@ -140,6 +166,7 @@ func (r *readStreamCache) release() {
 func (r *readStreamCache) releaseStream() {
 	r.once.Do(func() {
 		r.stream.CloseSend()
+		close(r.reqs)
 		r.state = stateClosed
 	})
 }
@@ -182,6 +209,7 @@ func (s *BlockStore) runAppendStreamRecv(ctx context.Context, append *appendStre
 				append.releaseCallbacks()
 				return
 			}
+			append.setState(stateClosed)
 			break
 		}
 		c, _ := append.callbacks.LoadAndDelete(res.Id)
@@ -191,6 +219,21 @@ func (s *BlockStore) runAppendStreamRecv(ctx context.Context, append *appendStre
 	}
 }
 
+func (s *BlockStore) runAppendStreamSend(ctx context.Context, append *appendStreamCache) {
+	for req := range append.reqs {
+		if err := append.stream.Send(req); err != nil {
+			log.Error(ctx, "append stream send failed", map[string]interface{}{
+				log.KeyError: err,
+			})
+			// close the current stream connection
+			append.releaseStream()
+			// reset new stream connections
+			s.connectAppendStream(ctx)
+			return
+		}
+	}
+}
+
 func (s *BlockStore) runReadStreamRecv(ctx context.Context, read *readStreamCache) {
 	for {
 		res, err := read.stream.Recv()
@@ -203,6 +246,7 @@ func (s *BlockStore) runReadStreamRecv(ctx context.Context, read *readStreamCach
 				read.releaseCallbacks()
 				return
 			}
+			read.setState(stateClosed)
 			break
 		}
 		c, _ := read.callbacks.LoadAndDelete(res.Id)
@@ -212,6 +256,21 @@ func (s *BlockStore) runReadStreamRecv(ctx context.Context, read *readStreamCach
 	}
 }
 
+func (s *BlockStore) runReadStreamSend(ctx context.Context, read *readStreamCache) {
+	for req := range read.reqs {
+		if err := read.stream.Send(req); err != nil {
+			log.Error(ctx, "read stream send failed", map[string]interface{}{
+				log.KeyError: err,
+			})
+			// close the current stream connection
+			read.releaseStream()
+			// reset new stream connections
+			s.connectReadStream(ctx)
+			return
+		}
+	}
+}
+
 func (s *BlockStore) connectAppendStream(ctx context.Context) (*appendStreamCache, error) {
 	s.appendMu.RLock()
 	if s.append != nil && s.append.isRunning() {
@@ -240,11 +299,13 @@ func (s *BlockStore) connectAppendStream(ctx context.Context) (*appendStreamCach
 
 	cache := &appendStreamCache{
 		stream:    stream,
+		reqs:      make(chan *segpb.AppendToBlockStreamRequest, 100),
 		state:     stateRunning,
 		callbacks: sync.Map{},
 	}
 	s.append = cache
 	go s.runAppendStreamRecv(ctx, cache)
+	go s.runAppendStreamSend(ctx, cache)
 	return cache, nil
 }
 
@@ -282,6 +343,7 @@ func (s *BlockStore) connectReadStream(ctx context.Context) (*readStreamCache, e
 	}
 	s.read = cache
 	go s.runReadStreamRecv(ctx, cache)
+	go s.runReadStreamSend(ctx, cache)
 	return cache, nil
 }
 
@@ -355,34 +417,14 @@ func (s *BlockStore) AppendManyStream(ctx context.Context, block uint64, events
 		close(donec)
 	}))
 
-	req := &segpb.AppendToBlockStreamRequest{
+	append.reqs <- &segpb.AppendToBlockStreamRequest{
 		Id:      opaqueID,
 		BlockId: block,
 		Events: &cepb.CloudEventBatch{
 			Events: eventpbs,
 		},
 	}
 
-	if err = append.stream.Send(req); err != nil {
-		log.Error(ctx, "append stream send failed", map[string]interface{}{
-			log.KeyError: err,
-		})
-		// close the current stream connection
-		append.releaseStream()
-		// reset new stream connections
-		s.connectAppendStream(ctx)
-		c, _ := append.callbacks.LoadAndDelete(opaqueID)
-		if c != nil {
-			c.(appendCallback)(&segpb.AppendToBlockStreamResponse{
-				Id:           opaqueID,
-				ResponseCode: errpb.ErrorCode_CLOSED,
-				ResponseMsg:  "append stream closed",
-				Offsets:      []int64{},
-			})
-		}
-		return nil, err
-	}
-
 	select {
 	case <-donec:
 	case <-_ctx.Done():
@@ -394,6 +436,8 @@ func (s *BlockStore) AppendManyStream(ctx context.Context, block uint64, events
 				ResponseMsg:  "append stream context canceled",
 				Offsets:      []int64{},
 			})
+		} else {
+			<-donec
 		}
 	}
 
@@ -475,34 +519,14 @@ func (s *BlockStore) ReadStream(
 		close(donec)
 	}))
 
-	req := &segpb.ReadFromBlockStreamRequest{
+	read.reqs <- &segpb.ReadFromBlockStreamRequest{
 		Id:                          opaqueID,
 		BlockId:                     block,
 		Offset:                      offset,
 		Number:                      int64(size),
 		PollingTimeoutInMillisecond: pollingTimeout,
 	}
 
-	if err = s.read.stream.Send(req); err != nil {
-		log.Error(ctx, "read stream send failed", map[string]interface{}{
-			log.KeyError: err,
-		})
-		read.releaseStream()
-		s.connectReadStream(ctx)
-		c, _ := read.callbacks.LoadAndDelete(opaqueID)
-		if c != nil {
-			c.(readCallback)(&segpb.ReadFromBlockStreamResponse{
-				Id:           opaqueID,
-				ResponseCode: errpb.ErrorCode_CLOSED,
-				ResponseMsg:  "read stream closed",
-				Events: &cepb.CloudEventBatch{
-					Events: []*cepb.CloudEvent{},
-				},
-			})
-		}
-		return []*ce.Event{}, err
-	}
-
 	select {
 	case <-donec:
 	case <-_ctx.Done():
@@ -516,6 +540,8 @@ func (s *BlockStore) ReadStream(
 					Events: []*cepb.CloudEvent{},
 				},
 			})
+		} else {
+			<-donec
 		}
 	}
 

client/pkg/api/client.go

@@ -33,6 +33,7 @@ type Eventbus interface {
 
 type BusWriter interface {
 	AppendOne(ctx context.Context, event *ce.Event, opts ...WriteOption) (eid string, err error)
+	// TODO(jiangkai): just keep one or make their difference more obvious.
 	AppendMany(ctx context.Context, events []*ce.Event, opts ...WriteOption) (eid []string, err error)
 	AppendBatch(ctx context.Context, events *cloudevents.CloudEventBatch, opts ...WriteOption) (err error)
 }

client/pkg/eventlog/eventlog.go

@@ -19,15 +19,15 @@ import (
 	// standard libraries.
 	"context"
 
-	"github.com/linkall-labs/vanus/proto/pkg/cloudevents"
-
 	// third-party libraries.
 	ce "github.com/cloudevents/sdk-go/v2"
 
 	// first-party libraries.
-	"github.com/linkall-labs/vanus/client/pkg/api"
+	"github.com/linkall-labs/vanus/proto/pkg/cloudevents"
 	segpb "github.com/linkall-labs/vanus/proto/pkg/segment"
+
 	// this project.
+	"github.com/linkall-labs/vanus/client/pkg/api"
 )
 
 const (
@@ -55,6 +55,7 @@ type LogWriter interface {
 
 	AppendMany(ctx context.Context, events *cloudevents.CloudEventBatch) (off int64, err error)
 
+	// TODO(jiangkai): unified event data format and maybe make stream as a option of writer is better
 	AppendManyStream(ctx context.Context, events []*ce.Event) ([]int64, error)
 }
 

client/pkg/eventlog/eventlog_impl.go

@@ -579,7 +579,7 @@ func (r *logReader) selectReadableSegment(ctx context.Context) (*segment, error)
 		segment = r.cur
 		if segment == nil { // double check
 			var err error
-			segment, err := r.elog.selectReadableSegment(ctx, r.pos)
+			segment, err = r.elog.selectReadableSegment(ctx, r.pos)
 			if errors.Is(err, errors.ErrOffsetOnEnd) {
 				r.elog.refreshReadableSegments(ctx)
 				segment, err = r.elog.selectReadableSegment(ctx, r.pos)