fixed update token for topic reader

Author: rekby

CHANGELOG.md

@@ -1,3 +1,4 @@
+* Fixed update token for topic reader
 * Marked sessions which creates from `database/sql` driver as supported server-side session balancing
 
 ## v3.37.7

config/config.go

@@ -35,6 +35,10 @@ type Config struct {
 	excludeGRPCCodesForPessimization []grpcCodes.Code
 }
 
+func (c Config) Credentials() credentials.Credentials {
+	return c.credentials
+}
+
 // ExcludeGRPCCodesForPessimization defines grpc codes for exclude its from pessimization trigger
 func (c Config) ExcludeGRPCCodesForPessimization() []grpcCodes.Code {
 	return c.excludeGRPCCodesForPessimization

connection.go

@@ -330,7 +330,7 @@ func (c *connection) Scripting() scripting.Client {
 // Notice: This API is EXPERIMENTAL and may be changed or removed in a later release.
 func (c *connection) Topic() topic.Client {
 	c.topicOnce.Init(func() closeFunc {
-		c.topic = topicclientinternal.New(c.balancer, c.topicOptions...)
+		c.topic = topicclientinternal.New(c.balancer, c.config.Credentials(), c.topicOptions...)
 		return c.topic.Close
 	})
 	return c.topic

internal/topic/topicclientinternal/client.go

@@ -6,6 +6,7 @@ import (
 	"github.com/ydb-platform/ydb-go-genproto/Ydb_Topic_V1"
 	"google.golang.org/grpc"
 
+	"github.com/ydb-platform/ydb-go-sdk/v3/credentials"
 	"github.com/ydb-platform/ydb-go-sdk/v3/internal/grpcwrapper/rawtopic"
 	"github.com/ydb-platform/ydb-go-sdk/v3/internal/grpcwrapper/rawtopic/rawtopiccommon"
 	"github.com/ydb-platform/ydb-go-sdk/v3/internal/grpcwrapper/rawydb"
@@ -18,11 +19,12 @@ import (
 
 type Client struct {
 	cfg                    topic.Config
+	cred                   credentials.Credentials
 	defaultOperationParams rawydb.OperationParams
 	rawClient              rawtopic.Client
 }
 
-func New(conn grpc.ClientConnInterface, opts ...topicoptions.TopicOption) *Client {
+func New(conn grpc.ClientConnInterface, cred credentials.Credentials, opts ...topicoptions.TopicOption) *Client {
 	rawClient := rawtopic.NewClient(Ydb_Topic_V1.NewTopicServiceClient(conn))
 
 	cfg := newTopicConfig(opts...)
@@ -32,6 +34,7 @@ func New(conn grpc.ClientConnInterface, opts ...topicoptions.TopicOption) *Clien
 
 	return &Client{
 		cfg:                    cfg,
+		cred:                   cred,
 		defaultOperationParams: defaultOperationParams,
 		rawClient:              rawClient,
 	}
@@ -160,7 +163,10 @@ func (c *Client) StartReader(
 		return c.rawClient.StreamRead(ctx)
 	}
 
-	defaultOpts := []topicoptions.ReaderOption{topicoptions.WithCommonConfig(c.cfg.Common)}
+	defaultOpts := []topicoptions.ReaderOption{
+		topicoptions.WithCommonConfig(c.cfg.Common),
+		topicreaderinternal.WithCredentials(c.cred),
+	}
 	opts = append(defaultOpts, opts...)
 
 	internalReader := topicreaderinternal.NewReader(connector, consumer, readSelectors, opts...)

internal/topic/topicreaderinternal/reader.go

@@ -7,6 +7,7 @@ import (
 	"sync"
 	"time"
 
+	"github.com/ydb-platform/ydb-go-sdk/v3/credentials"
 	"github.com/ydb-platform/ydb-go-sdk/v3/internal/backoff"
 	"github.com/ydb-platform/ydb-go-sdk/v3/internal/config"
 	"github.com/ydb-platform/ydb-go-sdk/v3/internal/grpcwrapper/rawtopic/rawtopicreader"
@@ -192,6 +193,15 @@ type ReaderConfig struct {
 // Notice: This API is EXPERIMENTAL and may be changed or removed in a later release.
 type PublicReaderOption func(cfg *ReaderConfig)
 
+func WithCredentials(cred credentials.Credentials) PublicReaderOption {
+	return func(cfg *ReaderConfig) {
+		if cred == nil {
+			cred = credentials.NewAnonymousCredentials()
+		}
+		cfg.Cred = cred
+	}
+}
+
 func convertNewParamsToStreamConfig(
 	consumer string,
 	readSelectors []PublicReadSelector,