fix linters

Author: rekby

internal/topic/topicreaderinternal/committer.go

@@ -54,7 +54,7 @@ type committer struct {
 	commits CommitRanges
 }
 
-func newCommitter(tracer trace.Topic, lifeContext context.Context, mode PublicCommitMode, send sendMessageToServerFunc) *committer {
+func newCommitter(tracer trace.Topic, lifeContext context.Context, mode PublicCommitMode, send sendMessageToServerFunc) *committer { //nolint:lll,revive
 	res := &committer{
 		mode:             mode,
 		clock:            clockwork.NewRealClock(),
@@ -141,7 +141,7 @@ func (c *committer) pushCommitsLoop(ctx context.Context) {
 
 		commits.optimize()
 
-		onDone := trace.TopicOnReaderStreamSendCommitMessage(c.tracer, &commits)
+		onDone := trace.TopicOnReaderSendCommitMessage(c.tracer, &commits)
 		err := sendCommitMessage(c.send, commits)
 		onDone(err)
 

internal/topic/topicreaderinternal/stream_reader_impl.go

@@ -164,7 +164,7 @@ func (r *topicStreamReaderImpl) ReadMessageBatch(
 	ctx context.Context,
 	opts ReadMessageBatchOptions,
 ) (batch *PublicBatch, err error) {
-	onDone := trace.TopicOnReaderStreamReadMessages(
+	onDone := trace.TopicOnReaderReadMessages(
 		r.cfg.Tracer,
 		ctx,
 		opts.MinCount,
@@ -330,7 +330,7 @@ func (r *topicStreamReaderImpl) Commit(ctx context.Context, commitRange commitRa
 		session = *commitRange.partitionSession
 	}
 
-	onDone := trace.TopicOnReaderStreamCommit(
+	onDone := trace.TopicOnReaderCommit(
 		r.cfg.Tracer,
 		ctx,
 		session.Topic,
@@ -371,7 +371,7 @@ func (r *topicStreamReaderImpl) checkCommitRange(commitRange commitRange) error
 func (r *topicStreamReaderImpl) send(msg rawtopicreader.ClientMessage) error {
 	err := r.stream.Send(msg)
 	if err != nil {
-		trace.TopicOnReaderStreamError(r.cfg.Tracer, r.readConnectionID, err)
+		trace.TopicOnReaderError(r.cfg.Tracer, r.readConnectionID, err)
 		_ = r.CloseWithError(r.ctx, err)
 	}
 	return err
@@ -406,7 +406,7 @@ func (r *topicStreamReaderImpl) setStarted() error {
 func (r *topicStreamReaderImpl) initSession() (err error) {
 	initMessage := r.cfg.initMessage()
 
-	onDone := trace.TopicOnReaderStreamInit(r.cfg.Tracer, r.readConnectionID, initMessage)
+	onDone := trace.TopicOnReaderInit(r.cfg.Tracer, r.readConnectionID, initMessage)
 	defer func() {
 		onDone(r.readConnectionID, err)
 	}()
@@ -453,9 +453,9 @@ func (r *topicStreamReaderImpl) readMessagesLoop(ctx context.Context) {
 	for {
 		serverMessage, err := r.stream.Recv()
 		if err != nil {
-			trace.TopicOnReaderStreamError(r.cfg.Tracer, r.readConnectionID, err)
+			trace.TopicOnReaderError(r.cfg.Tracer, r.readConnectionID, err)
 			if errors.Is(err, rawtopicreader.ErrUnexpectedMessageType) {
-				trace.TopicOnReaderStreamUnknownGrpcMessage(r.cfg.Tracer, r.readConnectionID, err)
+				trace.TopicOnReaderUnknownGrpcMessage(r.cfg.Tracer, r.readConnectionID, err)
 				// new messages can be added to protocol, it must be backward compatible to old programs
 				// and skip message is safe
 				continue
@@ -497,7 +497,7 @@ func (r *topicStreamReaderImpl) readMessagesLoop(ctx context.Context) {
 		case *rawtopicreader.UpdateTokenResponse:
 			// skip
 		default:
-			trace.TopicOnReaderStreamUnknownGrpcMessage(
+			trace.TopicOnReaderUnknownGrpcMessage(
 				r.cfg.Tracer,
 				r.readConnectionID,
 				xerrors.WithStackTrace(xerrors.Wrap(fmt.Errorf(
@@ -537,7 +537,7 @@ func (r *topicStreamReaderImpl) dataRequestLoop(ctx context.Context) {
 			}
 
 			resCapacity := r.addRestBufferBytes(sum)
-			trace.TopicOnReaderStreamSentDataRequest(r.cfg.Tracer, r.readConnectionID, sum, resCapacity)
+			trace.TopicOnReaderSentDataRequest(r.cfg.Tracer, r.readConnectionID, sum, resCapacity)
 			if err := r.sendDataRequest(sum); err != nil {
 				return
 			}
@@ -577,7 +577,7 @@ func (r *topicStreamReaderImpl) updateTokenLoop(ctx context.Context) {
 
 func (r *topicStreamReaderImpl) onReadResponse(msg *rawtopicreader.ReadResponse) (err error) {
 	resCapacity := r.addRestBufferBytes(-msg.BytesSize)
-	onDone := trace.TopicOnReaderStreamReceiveDataResponse(r.cfg.Tracer, r.readConnectionID, resCapacity, msg)
+	onDone := trace.TopicOnReaderReceiveDataResponse(r.cfg.Tracer, r.readConnectionID, resCapacity, msg)
 	defer func() {
 		onDone(err)
 	}()
@@ -624,7 +624,7 @@ func (r *topicStreamReaderImpl) onReadResponse(msg *rawtopicreader.ReadResponse)
 }
 
 func (r *topicStreamReaderImpl) CloseWithError(ctx context.Context, reason error) (closeErr error) {
-	onDone := trace.TopicOnReaderStreamClose(r.cfg.Tracer, r.readConnectionID, reason)
+	onDone := trace.TopicOnReaderClose(r.cfg.Tracer, r.readConnectionID, reason)
 	defer onDone(closeErr)
 
 	isFirstClose := false
@@ -672,7 +672,7 @@ func (r *topicStreamReaderImpl) onCommitResponse(msg *rawtopicreader.CommitOffse
 		}
 		partition.setCommittedOffset(commit.CommittedOffset)
 
-		trace.TopicOnReaderStreamCommittedNotify(
+		trace.TopicOnReaderCommittedNotify(
 			r.cfg.Tracer,
 			r.readConnectionID,
 			partition.Topic,
@@ -688,7 +688,7 @@ func (r *topicStreamReaderImpl) onCommitResponse(msg *rawtopicreader.CommitOffse
 }
 
 func (r *topicStreamReaderImpl) updateToken(ctx context.Context) {
-	onUpdateToken := trace.TopicOnReaderStreamUpdateToken(
+	onUpdateToken := trace.TopicOnReaderUpdateToken(
 		r.cfg.Tracer,
 		r.readConnectionID,
 	)
@@ -765,11 +765,7 @@ func (r *topicStreamReaderImpl) onStartPartitionSessionRequestFromBuffer(
 		respMessage.CommitOffset.FromInt64Pointer(commitOffset)
 	}
 
-	if err = r.send(respMessage); err != nil {
-		return err
-	}
-
-	return nil
+	return r.send(respMessage)
 }
 
 func (r *topicStreamReaderImpl) onStopPartitionSessionRequest(m *rawtopicreader.StopPartitionSessionRequest) error {

internal/topic/topicreaderinternal/stream_reader_impl_test.go

@@ -145,7 +145,7 @@ func TestStreamReaderImpl_OnPartitionCloseHandle(t *testing.T) {
 		readMessagesCtx, readMessagesCtxCancel := xcontext.WithErrCancel(context.Background())
 		committedOffset := int64(222)
 
-		e.reader.cfg.Tracer.OnReaderPartitionReadStopResponse = func(info trace.TopicReaderPartitionReadStopResponseStartInfo) func(doneInfo trace.TopicReaderPartitionReadStopResponseDoneInfo) {
+		e.reader.cfg.Tracer.OnReaderPartitionReadStopResponse = func(info trace.TopicReaderPartitionReadStopResponseStartInfo) func(doneInfo trace.TopicReaderPartitionReadStopResponseDoneInfo) { //nolint:lll
 			expected := trace.TopicReaderPartitionReadStopResponseStartInfo{
 				ReaderConnectionID: e.reader.readConnectionID,
 				PartitionContext:   e.partitionSession.ctx,

internal/topic/topicreaderinternal/trace_wrappers.go

@@ -6,8 +6,6 @@ package topic_test
 import (
 	"context"
 	"fmt"
-	"io"
-	"io/ioutil"
 	"runtime/pprof"
 	"sync/atomic"
 	"testing"
@@ -23,7 +21,6 @@ import (
 	"github.com/ydb-platform/ydb-go-sdk/v3/topic/topicreader"
 	"github.com/ydb-platform/ydb-go-sdk/v3/topic/topicsugar"
 	"github.com/ydb-platform/ydb-go-sdk/v3/topic/topictypes"
-	"github.com/ydb-platform/ydb-go-sdk/v3/trace"
 )
 
 func TestReadMessages(t *testing.T) {
@@ -188,26 +185,6 @@ WITH (
 }
 
 func createReader(t *testing.T, db ydb.Connection, opts ...topicoptions.ReaderOption) *topicreader.Reader {
-	js := func(v interface{ JSONData() io.Reader }) string {
-		data, err := ioutil.ReadAll(v.JSONData())
-		if err != nil {
-			t.Fatal(err)
-		}
-		return string(data)
-	}
-
-	tracer := trace.Topic{
-		OnReadStreamRawSent: func(info trace.OnReadStreamRawSentInfo) {
-			t.Logf("sent: %v %v\n%v", info.ClientMessage.Type(), info.Error, js(info.ClientMessage))
-		},
-		OnReadStreamRawReceived: func(info trace.OnReadStreamRawReceivedInfo) {
-			t.Logf("received: %v %v\n%v", info.ServerMessage.Type(), info.Error, js(info.ServerMessage))
-		},
-	}
-	_ = tracer
-
-	opts = append(opts[:len(opts):len(opts)], topicoptions.WithTracer(tracer))
-
 	topicPath := db.Name() + "/test/feed"
 	reader, err := db.Topic().StartReader("test", []topicoptions.ReadSelector{
 		{