fix linter

rekby · rekby · commit aaaf6902d6ae · 2024-07-24T18:53:22.000+03:00
diff --git a/internal/topic/topiclistenerinternal/event_handler.go b/internal/topic/topiclistenerinternal/event_handler.go
@@ -23,8 +23,8 @@ type EventHandler interface {
 	// Experimental: https://github.com/ydb-platform/ydb-go-sdk/blob/master/VERSIONING.md#experimental
 	OnReadMessages(ctx context.Context, event *PublicReadMessages) error
 
-	// OnStopPartitionSessionRequest called when the server send stop partition message. It means that no more OnReadMessages
-	// calls for the partition session.
+	// OnStopPartitionSessionRequest called when the server send stop partition message.
+	// It means that no more OnReadMessages calls for the partition session.
 	// You must call event.Confirm() for allow the server to stop the partition session (if event.Graceful=true).
 	// Confirm is optional for event.Graceful=false
 	// The method can be called twice: with event.Graceful=true, then event.Graceful=false.
@@ -49,7 +49,6 @@ type PublicStartPartitionSessionEvent struct {
 	PartitionSession topicreadercommon.PublicPartitionSession
 	CommittedOffset  int64
 	PartitionOffsets PublicOffsetsRange
-	resp             PublicStartPartitionSessionConfirm
 	respChan         chan PublicStartPartitionSessionConfirm
 }
 
@@ -77,6 +76,7 @@ type PublicStartPartitionSessionConfirm struct {
 // Experimental: https://github.com/ydb-platform/ydb-go-sdk/blob/master/VERSIONING.md#experimental
 func (c PublicStartPartitionSessionConfirm) WithReadOffet(val int64) PublicStartPartitionSessionConfirm {
 	c.readOffset = &val
+
 	return c
 }
 
@@ -85,6 +85,7 @@ func (c PublicStartPartitionSessionConfirm) WithReadOffet(val int64) PublicStart
 // Experimental: https://github.com/ydb-platform/ydb-go-sdk/blob/master/VERSIONING.md#experimental
 func (c PublicStartPartitionSessionConfirm) WithCommitOffset(val int64) PublicStartPartitionSessionConfirm {
 	c.CommitOffset = &val
+
 	return c
 }
 
diff --git a/internal/topic/topiclistenerinternal/stream_listener.go b/internal/topic/topiclistenerinternal/stream_listener.go
@@ -53,11 +53,13 @@ func newStreamListener(
 	res.initVars(sessionIDCounter)
 	if err := res.initStream(connectionCtx, client); err != nil {
 		res.closeWithTimeout(connectionCtx, err)
+
 		return nil, err
 	}
 
 	res.startBackground()
 	res.sendDataRequest(config.BufferSize)
+
 	return res, nil
 }
 
@@ -114,6 +116,7 @@ func (l *streamListener) initVars(sessionIDCounter *atomic.Int64) {
 	}
 }
 
+//nolint:funlen
 func (l *streamListener) initStream(ctx context.Context, client TopicClient) error {
 	streamCtx, streamClose := context.WithCancelCause(xcontext.ValueOnly(ctx))
 	l.streamClose = streamClose
@@ -150,20 +153,33 @@ func (l *streamListener) initStream(ctx context.Context, client TopicClient) err
 
 	resp, err := l.stream.Recv()
 	if err != nil {
-		return xerrors.WithStackTrace(fmt.Errorf("ydb: failed to receive init response for read stream in the listener: %w", err))
+		return xerrors.WithStackTrace(fmt.Errorf(
+			"ydb: failed to receive init response for read stream in the listener: %w",
+			err,
+		))
 	}
 
 	if status := resp.StatusData(); !status.Status.IsSuccess() {
-		// TODO: better handler status error
-		return xerrors.WithStackTrace(fmt.Errorf("ydb: received bad status on init the topic stream listener: %v (%v)", status.Status, status.Issues))
+		// wrap initialization error as operation status error - for handle with retrier
+		// https://github.com/ydb-platform/ydb-go-sdk/issues/1361
+		return xerrors.WithStackTrace(fmt.Errorf(
+			"ydb: received bad status on init the topic stream listener: %v (%v)",
+			status.Status,
+			status.Issues,
+		))
 	}
 
 	initResp, ok := resp.(*rawtopicreader.InitResponse)
 	if !ok {
-		return xerrors.WithStackTrace(fmt.Errorf("bad message type on session init: %v (%v)", resp, reflect.TypeOf(resp)))
+		return xerrors.WithStackTrace(fmt.Errorf(
+			"bad message type on session init: %v (%v)",
+			resp,
+			reflect.TypeOf(resp),
+		))
 	}
 
 	l.sessionID = initResp.SessionID
+
 	return nil
 }
 
@@ -178,13 +194,17 @@ func (l *streamListener) sendMessagesLoop(ctx context.Context) {
 			l.m.WithLock(func() {
 				messages = l.messagesToSend
 				if len(messages) > 0 {
-					l.messagesToSend = make([]rawtopicreader.ClientMessage, 0, len(messages)*2)
+					l.messagesToSend = make([]rawtopicreader.ClientMessage, 0, cap(messages))
 				}
 			})
 
 			for _, m := range messages {
 				if err := l.stream.Send(m); err != nil {
-					l.closeWithTimeout(ctx, xerrors.WithStackTrace(xerrors.Wrap(fmt.Errorf("ydb: failed send message by grpc to topic reader stream from listener: %w", err))))
+					l.closeWithTimeout(ctx, xerrors.WithStackTrace(xerrors.Wrap(fmt.Errorf(
+						"ydb: failed send message by grpc to topic reader stream from listener: %w",
+						err,
+					))))
+
 					return
 				}
 			}
@@ -203,6 +223,7 @@ func (l *streamListener) receiveMessagesLoop(ctx context.Context) {
 			l.closeWithTimeout(ctx, xerrors.WithStackTrace(
 				fmt.Errorf("ydb: failed read message from the stream in the topic reader listener: %w", err),
 			))
+
 			return
 		}
 
@@ -218,14 +239,17 @@ func (l *streamListener) onReceiveServerMessage(ctx context.Context, mess rawtop
 	case *rawtopicreader.StopPartitionSessionRequest:
 		err = l.onStopPartitionRequest(ctx, m)
 	case *rawtopicreader.ReadResponse:
-		err = l.onReadResponse(ctx, m)
+		err = l.onReadResponse(m)
 	}
 	if err != nil {
 		l.closeWithTimeout(ctx, err)
 	}
 }
 
-func (l *streamListener) onStartPartitionRequest(ctx context.Context, m *rawtopicreader.StartPartitionSessionRequest) error {
+func (l *streamListener) onStartPartitionRequest(
+	ctx context.Context,
+	m *rawtopicreader.StartPartitionSessionRequest,
+) error {
 	session := topicreadercommon.NewPartitionSession(
 		ctx,
 		m.PartitionSession.Path,
@@ -276,10 +300,14 @@ func (l *streamListener) onStartPartitionRequest(ctx context.Context, m *rawtopi
 	}
 
 	l.sendMessage(resp)
+
 	return nil
 }
 
-func (l *streamListener) onStopPartitionRequest(ctx context.Context, m *rawtopicreader.StopPartitionSessionRequest) error {
+func (l *streamListener) onStopPartitionRequest(
+	ctx context.Context,
+	m *rawtopicreader.StopPartitionSessionRequest,
+) error {
 	session, err := l.sessions.Get(m.PartitionSessionID)
 	if !m.Graceful && session == nil {
 		// stop partition may be received twice: graceful and force
@@ -331,10 +359,11 @@ func (l *streamListener) onStopPartitionRequest(ctx context.Context, m *rawtopic
 	if m.Graceful {
 		l.sendMessage(&rawtopicreader.StopPartitionSessionResponse{PartitionSessionID: session.StreamPartitionSessionID})
 	}
+
 	return nil
 }
 
-func (l *streamListener) onReadResponse(ctx context.Context, m *rawtopicreader.ReadResponse) error {
+func (l *streamListener) onReadResponse(m *rawtopicreader.ReadResponse) error {
 	batches, err := topicreadercommon.ReadRawBatchesToPublicBatches(m, l.sessions, l.cfg.Decoders)
 	if err != nil {
 		return err
@@ -349,6 +378,7 @@ func (l *streamListener) onReadResponse(ctx context.Context, m *rawtopicreader.R
 		}
 	}
 	l.sendDataRequest(m.BytesSize)
+
 	return nil
 }
 
diff --git a/internal/topic/topiclistenerinternal/stream_listener_fixtures_test.go b/internal/topic/topiclistenerinternal/stream_listener_fixtures_test.go
@@ -20,8 +20,10 @@ func StreamListener(e fixenv.Env) *streamListener {
 		}
 		l.handler = EventHandlerMock(e)
 		l.sessions = PartitionStorage(e)
+
 		return fixenv.NewGenericResult(l), nil
 	}
+
 	return fixenv.CacheResult(e, f)
 }
 
@@ -31,8 +33,10 @@ func PartitionStorage(e fixenv.Env) *topicreadercommon.PartitionSessionStorage {
 		if err := storage.Add(PartitionSession(e)); err != nil {
 			return nil, err
 		}
+
 		return fixenv.NewGenericResult(storage), nil
 	}
+
 	return fixenv.CacheResult(e, f)
 }
 
@@ -49,29 +53,36 @@ func PartitionSession(e fixenv.Env) *topicreadercommon.PartitionSession {
 			0,
 		)), nil
 	}
+
 	return fixenv.CacheResult(e, f)
 }
 
 func MockController(e fixenv.Env) *gomock.Controller {
 	f := func() (*fixenv.GenericResult[*gomock.Controller], error) {
 		mc := gomock.NewController(e.T().(gomock.TestReporter))
+
 		return fixenv.NewGenericResult(mc), nil
 	}
+
 	return fixenv.CacheResult(e, f)
 }
 
 func StreamMock(e fixenv.Env) *rawtopicreadermock.MockTopicReaderStreamInterface {
 	f := func() (*fixenv.GenericResult[*rawtopicreadermock.MockTopicReaderStreamInterface], error) {
 		m := rawtopicreadermock.NewMockTopicReaderStreamInterface(MockController(e))
+
 		return fixenv.NewGenericResult(m), nil
 	}
+
 	return fixenv.CacheResult(e, f)
 }
 
 func EventHandlerMock(e fixenv.Env) *MockEventHandler {
 	f := func() (*fixenv.GenericResult[*MockEventHandler], error) {
 		m := NewMockEventHandler(MockController(e))
+
 		return fixenv.NewGenericResult(m), nil
 	}
+
 	return fixenv.CacheResult(e, f)
 }
diff --git a/internal/topic/topiclistenerinternal/stream_listener_test.go b/internal/topic/topiclistenerinternal/stream_listener_test.go
@@ -34,7 +34,8 @@ func TestStreamListener_OnReceiveServerMessage(t *testing.T) {
 		EventHandlerMock(e).EXPECT().OnReadMessages(PartitionSession(e).Context(), gomock.Any()).
 			DoAndReturn(func(ctx context.Context, event *PublicReadMessages) error {
 				require.Equal(t, PartitionSession(e).ClientPartitionSessionID, event.PartitionSession.PartitionSessionID)
-				require.Equal(t, event.Batch.Messages[0].SeqNo, seqNo)
+				require.Equal(t, seqNo, event.Batch.Messages[0].SeqNo)
+
 				return nil
 			})
 
@@ -92,6 +93,7 @@ func TestStreamListener_OnReceiveServerMessage(t *testing.T) {
 				WithReadOffet(respReadOffset).
 				WithCommitOffset(respCommitOffset),
 			)
+
 			return nil
 		})
 
@@ -142,6 +144,7 @@ func TestStreamListener_OnReceiveServerMessage(t *testing.T) {
 			require.True(t, event.Graceful)
 			require.Equal(t, int64(5), event.CommittedOffset)
 			event.Confirm()
+
 			return nil
 		})
 
@@ -175,6 +178,7 @@ func TestStreamListener_CloseSessionsOnCloseListener(t *testing.T) {
 		require.False(t, event.Graceful)
 		require.Equal(t, PartitionSession(e).CommittedOffset().ToInt64(), event.CommittedOffset)
 		event.Confirm()
+
 		return nil
 	})
 	require.NoError(t, StreamListener(e).Close(sf.Context(e), errors.New("test")))
diff --git a/internal/topic/topiclistenerinternal/topic_listener_reconnector.go b/internal/topic/topiclistenerinternal/topic_listener_reconnector.go
@@ -57,7 +57,13 @@ func (lr *TopicListenerReconnector) Close(ctx context.Context, reason error) err
 }
 
 func (lr *TopicListenerReconnector) connect(connectionCtx context.Context) {
-	lr.streamListener, lr.connectionResult = newStreamListener(connectionCtx, lr.client, lr.handler, lr.streamConfig, &lr.connectionIDCounter)
+	lr.streamListener, lr.connectionResult = newStreamListener(
+		connectionCtx,
+		lr.client,
+		lr.handler,
+		lr.streamConfig,
+		&lr.connectionIDCounter,
+	)
 	close(lr.connectionCompleted)
 }
 
@@ -85,6 +91,7 @@ func (lr *TopicListenerReconnector) WaitStop(ctx context.Context) error {
 		if errors.Is(err, ErrUserCloseTopic) {
 			return nil
 		}
+
 		return err
 	}
 }
diff --git a/internal/topic/topicreaderinternal/callback_reader.go b/internal/topic/topicreaderinternal/callback_reader.go
diff --git a/topic/topiclistener/topic_listener.go b/topic/topiclistener/topic_listener.go
@@ -12,7 +12,11 @@ type TopicListener struct {
 	listenerReconnector *topiclistenerinternal.TopicListenerReconnector
 }
 
-func NewTopicListener(client *rawtopic.Client, config *topiclistenerinternal.StreamListenerConfig, handler EventHandler) (*TopicListener, error) {
+func NewTopicListener(
+	client *rawtopic.Client, //nolint:interfacer
+	config *topiclistenerinternal.StreamListenerConfig,
+	handler EventHandler,
+) (*TopicListener, error) {
 	reconnector, err := topiclistenerinternal.NewTopicListenerReconnector(client, config, handler)
 	if err != nil {
 		return nil, err
diff --git a/topic/topicoptions/topicoptions_reader.go b/topic/topicoptions/topicoptions_reader.go
@@ -25,9 +25,6 @@ func ReadTopic(path string) ReadSelectors {
 // ReaderOption options for topic reader
 type ReaderOption = topicreaderinternal.PublicReaderOption
 
-// CallbackReaderOption options for topic callback reader
-type CallbackReaderOption = topicreaderinternal.PublicCallbackReaderOption
-
 // WithReaderOperationTimeout
 //
 // Experimental: https://github.com/ydb-platform/ydb-go-sdk/blob/master/VERSIONING.md#experimental

Original file line number	Diff line number	Diff line change
`@@ -20,8 +20,10 @@ func StreamListener(e fixenv.Env) *streamListener {`
`20`	`20`	`}`
`21`	`21`	`l.handler = EventHandlerMock(e)`
`22`	`22`	`l.sessions = PartitionStorage(e)`
	`23`	`+`
`23`	`24`	`return fixenv.NewGenericResult(l), nil`
`24`	`25`	`}`
	`26`	`+`
`25`	`27`	`return fixenv.CacheResult(e, f)`
`26`	`28`	`}`
`27`	`29`
`@@ -31,8 +33,10 @@ func PartitionStorage(e fixenv.Env) *topicreadercommon.PartitionSessionStorage {`
`31`	`33`	`if err := storage.Add(PartitionSession(e)); err != nil {`
`32`	`34`	`return nil, err`
`33`	`35`	`}`
	`36`	`+`
`34`	`37`	`return fixenv.NewGenericResult(storage), nil`
`35`	`38`	`}`
	`39`	`+`
`36`	`40`	`return fixenv.CacheResult(e, f)`
`37`	`41`	`}`
`38`	`42`
`@@ -49,29 +53,36 @@ func PartitionSession(e fixenv.Env) *topicreadercommon.PartitionSession {`
`49`	`53`	`0,`
`50`	`54`	`)), nil`
`51`	`55`	`}`
	`56`	`+`
`52`	`57`	`return fixenv.CacheResult(e, f)`
`53`	`58`	`}`
`54`	`59`
`55`	`60`	`func MockController(e fixenv.Env) *gomock.Controller {`
`56`	`61`	`f := func() (fixenv.GenericResult[gomock.Controller], error) {`
`57`	`62`	`mc := gomock.NewController(e.T().(gomock.TestReporter))`
	`63`	`+`
`58`	`64`	`return fixenv.NewGenericResult(mc), nil`
`59`	`65`	`}`
	`66`	`+`
`60`	`67`	`return fixenv.CacheResult(e, f)`
`61`	`68`	`}`
`62`	`69`
`63`	`70`	`func StreamMock(e fixenv.Env) *rawtopicreadermock.MockTopicReaderStreamInterface {`
`64`	`71`	`f := func() (fixenv.GenericResult[rawtopicreadermock.MockTopicReaderStreamInterface], error) {`
`65`	`72`	`m := rawtopicreadermock.NewMockTopicReaderStreamInterface(MockController(e))`
	`73`	`+`
`66`	`74`	`return fixenv.NewGenericResult(m), nil`
`67`	`75`	`}`
	`76`	`+`
`68`	`77`	`return fixenv.CacheResult(e, f)`
`69`	`78`	`}`
`70`	`79`
`71`	`80`	`func EventHandlerMock(e fixenv.Env) *MockEventHandler {`
`72`	`81`	`f := func() (fixenv.GenericResult[MockEventHandler], error) {`
`73`	`82`	`m := NewMockEventHandler(MockController(e))`
	`83`	`+`
`74`	`84`	`return fixenv.NewGenericResult(m), nil`
`75`	`85`	`}`
	`86`	`+`
`76`	`87`	`return fixenv.CacheResult(e, f)`
`77`	`88`	`}`
Original file line number	Diff line number	Diff line change
`@@ -57,7 +57,13 @@ func (lr *TopicListenerReconnector) Close(ctx context.Context, reason error) err`
`57`	`57`	`}`
`58`	`58`
`59`	`59`	`func (lr *TopicListenerReconnector) connect(connectionCtx context.Context) {`
`60`		`- lr.streamListener, lr.connectionResult = newStreamListener(connectionCtx, lr.client, lr.handler, lr.streamConfig, &lr.connectionIDCounter)`
	`60`	`+ lr.streamListener, lr.connectionResult = newStreamListener(`
	`61`	`+ connectionCtx,`
	`62`	`+ lr.client,`
	`63`	`+ lr.handler,`
	`64`	`+ lr.streamConfig,`
	`65`	`+ &lr.connectionIDCounter,`
	`66`	`+ )`
`61`	`67`	`close(lr.connectionCompleted)`
`62`	`68`	`}`
`63`	`69`
`@@ -85,6 +91,7 @@ func (lr *TopicListenerReconnector) WaitStop(ctx context.Context) error {`
`85`	`91`	`if errors.Is(err, ErrUserCloseTopic) {`
`86`	`92`	`return nil`
`87`	`93`	`}`
	`94`	`+`
`88`	`95`	`return err`
`89`	`96`	`}`
`90`	`97`	`}`