fix linter

Author: rekby

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
 }
 

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
 }
 

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)
 }

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")))

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
 	}
 }

internal/topic/topicreaderinternal/callback_reader.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

topic/topiclistener/topic_listener.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