fix generation and linter

Author: rekby

internal/topic/topicreadercommon/batch.go

@@ -178,7 +178,7 @@ func splitBytesByMessagesInBatches(batches []*PublicBatch, totalBytesCount int)
 	return nil
 }
 
-func BatchAppend(original *PublicBatch, appended *PublicBatch) (*PublicBatch, error) {
+func BatchAppend(original, appended *PublicBatch) (*PublicBatch, error) {
 	var res *PublicBatch
 	if original == nil {
 		res = &PublicBatch{}

internal/topic/topicreadercommon/split_batches.go

@@ -2,7 +2,11 @@ package topicreadercommon
 
 import "github.com/ydb-platform/ydb-go-sdk/v3/internal/grpcwrapper/rawtopic/rawtopicreader"
 
-func ReadRawBatchesToPublicBatches(msg *rawtopicreader.ReadResponse, sessions *PartitionSessionStorage, decoders DecoderMap) ([]*PublicBatch, error) {
+func ReadRawBatchesToPublicBatches(
+	msg *rawtopicreader.ReadResponse,
+	sessions *PartitionSessionStorage,
+	decoders DecoderMap,
+) ([]*PublicBatch, error) {
 	batchesCount := 0
 	for i := range msg.PartitionData {
 		batchesCount += len(msg.PartitionData[i].Batches)

internal/topic/topicreaderinternal/batcher.go

@@ -59,7 +59,10 @@ func (b *batcher) PushBatches(batches ...*topicreadercommon.PublicBatch) error {
 	}
 
 	for _, batch := range batches {
-		if err := b.addNeedLock(topicreadercommon.GetCommitRange(batch).PartitionSession, newBatcherItemBatch(batch)); err != nil {
+		if err := b.addNeedLock(
+			topicreadercommon.GetCommitRange(batch).PartitionSession,
+			newBatcherItemBatch(batch),
+		); err != nil {
 			return err
 		}
 	}
@@ -135,7 +138,10 @@ func (o batcherGetOptions) cutBatchItemsHead(items batcherMessageOrderItems) (
 	return items[0], items[1:], true
 }
 
-func (o batcherGetOptions) splitBatch(batch *topicreadercommon.PublicBatch) (head, rest *topicreadercommon.PublicBatch, ok bool) {
+func (o batcherGetOptions) splitBatch(batch *topicreadercommon.PublicBatch) (
+	head, rest *topicreadercommon.PublicBatch,
+	ok bool,
+) {
 	notFound := func() (*topicreadercommon.PublicBatch, *topicreadercommon.PublicBatch, bool) {
 		return nil, nil, false
 	}

internal/topic/topicreaderinternal/batcher_test.go

@@ -103,7 +103,10 @@ func TestBatcher_PushRawMessage(t *testing.T) {
 		expectedMap := batcherMessagesMap{session: batcherMessageOrderItems{
 			newBatcherItemBatch(batch1),
 			newBatcherItemRawMessage(m),
-			newBatcherItemBatch(mustNewBatch(session, []*topicreadercommon.PublicMessage{{WrittenAt: testTime(2)}, {WrittenAt: testTime(3)}})),
+			newBatcherItemBatch(mustNewBatch(
+				session,
+				[]*topicreadercommon.PublicMessage{{WrittenAt: testTime(2)}, {WrittenAt: testTime(3)}},
+			)),
 		}}
 		require.Equal(t, expectedMap, b.messages)
 	})
@@ -219,7 +222,10 @@ func TestBatcher_Pop(t *testing.T) {
 		b := newBatcher()
 		m := &rawtopicreader.StopPartitionSessionRequest{PartitionSessionID: 1}
 
-		require.NoError(t, b.PushBatches(mustNewBatch(session1, []*topicreadercommon.PublicMessage{{WrittenAt: testTime(1)}})))
+		require.NoError(t, b.PushBatches(mustNewBatch(
+			session1,
+			[]*topicreadercommon.PublicMessage{{WrittenAt: testTime(1)}},
+		)))
 		require.NoError(t, b.PushRawMessage(session2, m))
 
 		res, err := b.Pop(context.Background(), batcherGetOptions{})

internal/topic/topicreaderinternal/committer.go

@@ -263,7 +263,10 @@ type commitWaiter struct {
 	Committed empty.Chan
 }
 
-func (w *commitWaiter) checkCondition(session *topicreadercommon.PartitionSession, offset rawtopicreader.Offset) (finished bool) {
+func (w *commitWaiter) checkCondition(
+	session *topicreadercommon.PartitionSession,
+	offset rawtopicreader.Offset,
+) (finished bool) {
 	return session == w.Session && offset >= w.EndOffset
 }
 

internal/topic/topicreaderinternal/reader.go

@@ -140,7 +140,13 @@ func (r *Reader) ReadMessage(ctx context.Context) (*topicreadercommon.PublicMess
 
 // ReadMessageBatch read batch of messages.
 // Batch is collection of messages, which can be atomically committed
-func (r *Reader) ReadMessageBatch(ctx context.Context, opts ...PublicReadBatchOption) (batch *topicreadercommon.PublicBatch, err error) {
+func (r *Reader) ReadMessageBatch(
+	ctx context.Context,
+	opts ...PublicReadBatchOption,
+) (
+	batch *topicreadercommon.PublicBatch,
+	err error,
+) {
 	readOptions := r.defaultBatchConfig.clone()
 
 	for _, opt := range opts {

internal/topic/topicreaderinternal/stream_reader_impl.go

@@ -461,7 +461,8 @@ func (r *topicStreamReaderImpl) initSession() (err error) {
 	}
 
 	if status := resp.StatusData(); !status.Status.IsSuccess() {
-		// TODO: better handler status error
+		// Need wrap status to common ydb operational error
+		// https://github.com/ydb-platform/ydb-go-sdk/issues/1361
 		return xerrors.WithStackTrace(fmt.Errorf("bad status on initial error: %v (%v)", status.Status, status.Issues))
 	}
 

internal/topic/topicreaderinternal/stream_reconnector.go

@@ -77,7 +77,13 @@ func newReaderReconnector(
 	return res
 }
 
-func (r *readerReconnector) ReadMessageBatch(ctx context.Context, opts ReadMessageBatchOptions) (*topicreadercommon.PublicBatch, error) {
+func (r *readerReconnector) ReadMessageBatch(
+	ctx context.Context,
+	opts ReadMessageBatchOptions,
+) (
+	*topicreadercommon.PublicBatch,
+	error,
+) {
 	if ctx.Err() != nil {
 		return nil, ctx.Err()
 	}
@@ -127,7 +133,10 @@ func (r *readerReconnector) ReadMessageBatch(ctx context.Context, opts ReadMessa
 	}
 }
 
-func (r *readerReconnector) Commit(ctx context.Context, commitRange topicreadercommon.CommitRange) error {
+func (r *readerReconnector) Commit(
+	ctx context.Context,
+	commitRange topicreadercommon.CommitRange,
+) error {
 	stream, err := r.stream(ctx)
 	if err != nil {
 		return err