dev: added ValueTask<string?> GetAuthInfoAsync() in ICredentialProvider

CHANGELOG.md

@@ -1,3 +1,4 @@
+- Dev: added `ValueTask<string?> GetAuthInfoAsync()` in ICredentialProvider.
 - Feat: `Writer.DisposeAsync()` waits for all in-flight messages to complete.
 - Feat: `Reader.DisposeAsync()` waits for all pending commits to be completed.
 - **Breaking Change**: `IReader` now implements `IAsyncDisposable` instead of `IDisposable`.  

examples/src/BasicExample/ReadTable.cs

@@ -6,7 +6,7 @@ internal partial class BasicExample
 {
     private async Task ReadTable()
     {
-        var readStream = Client.ReadTable(
+        var readStream = await Client.ReadTable(
             FullTablePath("seasons"),
             new ReadTableSettings
             {

examples/src/BasicExample/ScanQuery.cs

@@ -18,7 +18,7 @@ FROM episodes
                 ORDER BY series_id, season_id;
             ";
 
-        var scanStream = Client.ExecuteScanQuery(
+        var scanStream = await Client.ExecuteScanQuery(
             query,
             new Dictionary<string, YdbValue>
             {

examples/src/QueryExample/QueryExample.cs

@@ -231,7 +231,7 @@ private async Task StreamSelect()
 
         await Client.DoTx(async tx =>
         {
-            await foreach (var part in tx.Stream(query, commit: true))
+            await foreach (var part in await tx.Stream(query, commit: true))
             {
                 foreach (var row in part.ResultSet!.Rows)
                 {
@@ -316,7 +316,7 @@ private async Task ReadAllResultSets()
         var resultSets = await Client.DoTx(async tx =>
         {
             var resultSets = new List<Value.ResultSet>();
-            await foreach (var resultSet in tx.Stream(query, commit: true))
+            await foreach (var resultSet in await tx.Stream(query, commit: true))
             {
                 resultSets.Add(resultSet.ResultSet!);
             }

src/Ydb.Sdk/src/Ado/YdbCommand.cs

@@ -223,9 +223,12 @@ protected override async Task<DbDataReader> ExecuteDbDataReaderAsync(CommandBeha
             throw new InvalidOperationException("Transaction mismatched! (Maybe using another connection)");
         }
 
-        var ydbDataReader = await YdbDataReader.CreateYdbDataReader(YdbConnection.Session.ExecuteQuery(
-                preparedSql.ToString(), ydbParameters, execSettings, transaction?.TransactionControl),
-            YdbConnection.Session.OnStatus, transaction);
+        var ydbDataReader = await YdbDataReader.CreateYdbDataReader(
+            await YdbConnection.Session.ExecuteQuery(
+                preparedSql.ToString(), ydbParameters, execSettings, transaction?.TransactionControl
+            ),
+            YdbConnection.Session.OnStatus, transaction
+        );
 
         YdbConnection.LastReader = ydbDataReader;
         YdbConnection.LastCommand = CommandText;

src/Ydb.Sdk/src/Auth/ICredentialsProvider.cs

@@ -4,8 +4,14 @@ namespace Ydb.Sdk.Auth;
 
 public interface ICredentialsProvider
 {
+    // For removal in 1.*
     string? GetAuthInfo();
 
+    ValueTask<string?> GetAuthInfoAsync()
+    {
+        return ValueTask.FromResult(GetAuthInfo());
+    }
+
     Task ProvideAuthClient(AuthClient authClient)
     {
         return Task.CompletedTask;

src/Ydb.Sdk/src/Driver.cs

@@ -129,7 +129,7 @@ private async Task<Status> DiscoverEndpoints()
             TransportTimeout = Config.EndpointDiscoveryTimeout
         };
 
-        var options = GetCallOptions(requestSettings);
+        var options = await GetCallOptions(requestSettings);
         options.Headers?.Add(Metadata.RpcSdkInfoHeader, _sdkInfo);
 
         var response = await client.ListEndpointsAsync(

src/Ydb.Sdk/src/IDriver.cs

@@ -14,14 +14,14 @@ public Task<TResponse> UnaryCall<TRequest, TResponse>(
         where TRequest : class
         where TResponse : class;
 
-    public ServerStream<TResponse> ServerStreamCall<TRequest, TResponse>(
+    public ValueTask<ServerStream<TResponse>> ServerStreamCall<TRequest, TResponse>(
         Method<TRequest, TResponse> method,
         TRequest request,
         GrpcRequestSettings settings)
         where TRequest : class
         where TResponse : class;
 
-    public IBidirectionalStream<TRequest, TResponse> BidirectionalStreamCall<TRequest, TResponse>(
+    public ValueTask<IBidirectionalStream<TRequest, TResponse>> BidirectionalStreamCall<TRequest, TResponse>(
         Method<TRequest, TResponse> method,
         GrpcRequestSettings settings)
         where TRequest : class
@@ -38,7 +38,7 @@ public interface IBidirectionalStream<in TRequest, out TResponse> : IDisposable
 
     public TResponse Current { get; }
 
-    public string? AuthToken { get; }
+    public ValueTask<string?> AuthToken { get; }
 
     public Task RequestStreamComplete();
 }
@@ -74,7 +74,7 @@ public async Task<TResponse> UnaryCall<TRequest, TResponse>(
             using var call = callInvoker.AsyncUnaryCall(
                 method: method,
                 host: null,
-                options: GetCallOptions(settings),
+                options: await GetCallOptions(settings),
                 request: request
             );
 
@@ -90,7 +90,7 @@ public async Task<TResponse> UnaryCall<TRequest, TResponse>(
         }
     }
 
-    public ServerStream<TResponse> ServerStreamCall<TRequest, TResponse>(
+    public async ValueTask<ServerStream<TResponse>> ServerStreamCall<TRequest, TResponse>(
         Method<TRequest, TResponse> method,
         TRequest request,
         GrpcRequestSettings settings)
@@ -103,13 +103,13 @@ public ServerStream<TResponse> ServerStreamCall<TRequest, TResponse>(
         var call = callInvoker.AsyncServerStreamingCall(
             method: method,
             host: null,
-            options: GetCallOptions(settings),
+            options: await GetCallOptions(settings),
             request: request);
 
         return new ServerStream<TResponse>(call, e => { OnRpcError(endpoint, e); });
     }
 
-    public IBidirectionalStream<TRequest, TResponse> BidirectionalStreamCall<TRequest, TResponse>(
+    public async ValueTask<IBidirectionalStream<TRequest, TResponse>> BidirectionalStreamCall<TRequest, TResponse>(
         Method<TRequest, TResponse> method,
         GrpcRequestSettings settings)
         where TRequest : class
@@ -121,7 +121,7 @@ public IBidirectionalStream<TRequest, TResponse> BidirectionalStreamCall<TReques
         var call = callInvoker.AsyncDuplexStreamingCall(
             method: method,
             host: null,
-            options: GetCallOptions(settings));
+            options: await GetCallOptions(settings));
 
         return new BidirectionalStream<TRequest, TResponse>(
             call,
@@ -133,14 +133,14 @@ public IBidirectionalStream<TRequest, TResponse> BidirectionalStreamCall<TReques
 
     protected abstract void OnRpcError(string endpoint, RpcException e);
 
-    protected CallOptions GetCallOptions(GrpcRequestSettings settings)
+    protected async ValueTask<CallOptions> GetCallOptions(GrpcRequestSettings settings)
     {
         var meta = new Grpc.Core.Metadata
         {
             { Metadata.RpcDatabaseHeader, Config.Database }
         };
 
-        var authInfo = Config.Credentials.GetAuthInfo();
+        var authInfo = await Config.Credentials.GetAuthInfoAsync();
         if (authInfo != null)
         {
             meta.Add(Metadata.RpcAuthHeader, authInfo);
@@ -268,7 +268,7 @@ public async ValueTask<bool> MoveNextAsync()
 
     public TResponse Current => _stream.ResponseStream.Current;
 
-    public string? AuthToken => _credentialsProvider.GetAuthInfo();
+    public ValueTask<string?> AuthToken => _credentialsProvider.GetAuthInfoAsync();
 
     public async Task RequestStreamComplete()
     {

src/Ydb.Sdk/src/Services/Query/QueryClient.cs

@@ -37,8 +37,9 @@ public Task<T> Stream<T>(string query, Func<ExecuteQueryStream, Task<T>> onStrea
         Dictionary<string, YdbValue>? parameters = null, TxMode txMode = TxMode.NoTx,
         ExecuteQuerySettings? settings = null)
     {
-        return _sessionPool.ExecOnSession(session => onStream(new ExecuteQueryStream(
-            session.ExecuteQuery(query, parameters, settings, txMode.TransactionControl()))));
+        return _sessionPool.ExecOnSession(async session => await onStream(new ExecuteQueryStream(
+            await session.ExecuteQuery(query, parameters, settings, txMode.TransactionControl())))
+        );
     }
 
     public Task Stream(string query, Func<ExecuteQueryStream, Task> onStream,

src/Ydb.Sdk/src/Services/Query/QueryTx.cs

@@ -27,17 +27,18 @@ internal QueryTx(Session session, TxMode txMode)
         _txMode = txMode;
     }
 
-    public ExecuteQueryStream Stream(string query, Dictionary<string, YdbValue>? parameters = null,
+    public async ValueTask<ExecuteQueryStream> Stream(string query, Dictionary<string, YdbValue>? parameters = null,
         bool commit = false, ExecuteQuerySettings? settings = null)
     {
-        return new ExecuteQueryStream(_session.ExecuteQuery(query, parameters, settings, TxControl(commit)),
-            txId => TxId = txId);
+        return new ExecuteQueryStream(
+            await _session.ExecuteQuery(query, parameters, settings, TxControl(commit)), txId => TxId = txId
+        );
     }
 
     public async Task<IReadOnlyList<Value.ResultSet.Row>> ReadAllRows(string query,
         Dictionary<string, YdbValue>? parameters = null, bool commit = false, ExecuteQuerySettings? settings = null)
     {
-        await using var stream = Stream(query, parameters, commit, settings);
+        await using var stream = await Stream(query, parameters, commit, settings);
         List<Value.ResultSet.Row> rows = new();
 
         await foreach (var part in stream)
@@ -64,7 +65,7 @@ public ExecuteQueryStream Stream(string query, Dictionary<string, YdbValue>? par
     public async Task Exec(string query, Dictionary<string, YdbValue>? parameters = null,
         ExecuteQuerySettings? settings = null, bool commit = false)
     {
-        await using var stream = Stream(query, parameters, commit, settings);
+        await using var stream = await Stream(query, parameters, commit, settings);
         await stream.MoveNextAsync();
     }
 

src/Ydb.Sdk/src/Services/Query/SessionPool.cs

@@ -57,7 +57,7 @@ protected override async Task<Session> CreateSession()
         {
             try
             {
-                await using var stream = _driver.ServerStreamCall(
+                await using var stream = await _driver.ServerStreamCall(
                     QueryService.AttachSessionMethod,
                     new AttachSessionRequest { SessionId = sessionId },
                     new GrpcRequestSettings { NodeId = nodeId }
@@ -136,7 +136,7 @@ ILogger<Session> logger
         Driver = driver;
     }
 
-    internal ServerStream<ExecuteQueryResponsePart> ExecuteQuery(
+    internal ValueTask<ServerStream<ExecuteQueryResponsePart>> ExecuteQuery(
         string query,
         Dictionary<string, YdbValue>? parameters,
         ExecuteQuerySettings? settings,

src/Ydb.Sdk/src/Services/Table/ExecuteScanQuery.cs

@@ -57,7 +57,7 @@ protected override ExecuteScanQueryPart MakeResponse(ExecuteScanQueryPartialResp
 
 public partial class TableClient
 {
-    public ExecuteScanQueryStream ExecuteScanQuery(
+    public async ValueTask<ExecuteScanQueryStream> ExecuteScanQuery(
         string query,
         IReadOnlyDictionary<string, YdbValue> parameters,
         ExecuteScanQuerySettings? settings = null)
@@ -75,7 +75,7 @@ public ExecuteScanQueryStream ExecuteScanQuery(
 
         request.Parameters.Add(parameters.ToDictionary(p => p.Key, p => p.Value.GetProto()));
 
-        var streamIterator = _driver.ServerStreamCall(
+        var streamIterator = await _driver.ServerStreamCall(
             method: TableService.StreamExecuteScanQueryMethod,
             request: request,
             settings: settings
@@ -84,7 +84,7 @@ public ExecuteScanQueryStream ExecuteScanQuery(
         return new ExecuteScanQueryStream(streamIterator);
     }
 
-    public ExecuteScanQueryStream ExecuteScanQuery(
+    public ValueTask<ExecuteScanQueryStream> ExecuteScanQuery(
         string query,
         ExecuteScanQuerySettings? settings = null)
     {

src/Ydb.Sdk/src/Services/Table/ReadTable.cs

@@ -62,7 +62,7 @@ protected override ReadTablePart MakeResponse(ReadTableResponse protoResponse)
 
 public partial class TableClient
 {
-    public ReadTableStream ReadTable(string tablePath, ReadTableSettings? settings = null)
+    public async ValueTask<ReadTableStream> ReadTable(string tablePath, ReadTableSettings? settings = null)
     {
         settings ??= new ReadTableSettings();
 
@@ -74,7 +74,7 @@ public ReadTableStream ReadTable(string tablePath, ReadTableSettings? settings =
             Ordered = settings.Ordered
         };
 
-        var streamIterator = _driver.ServerStreamCall(
+        var streamIterator = await _driver.ServerStreamCall(
             method: TableService.StreamReadTableMethod,
             request: request,
             settings: settings

src/Ydb.Sdk/src/Services/Topic/Reader/Reader.cs

@@ -103,7 +103,8 @@ private async Task Initialize()
 
             _logger.LogInformation("Reader session initialization started. ReaderConfig: {ReaderConfig}", _config);
 
-            var stream = _driver.BidirectionalStreamCall(TopicService.StreamReadMethod, _readerGrpcRequestSettings);
+            var stream =
+                await _driver.BidirectionalStreamCall(TopicService.StreamReadMethod, _readerGrpcRequestSettings);
 
             var initRequest = new StreamReadMessage.Types.InitRequest();
             if (_config.ConsumerName != null)
@@ -191,6 +192,7 @@ await stream.Write(new MessageFromClient
                 stream,
                 initResponse.SessionId,
                 Initialize,
+                await stream.AuthToken,
                 _logger,
                 _receivedMessagesChannel.Writer,
                 _deserializer
@@ -270,14 +272,16 @@ public ReaderSession(
         ReaderStream stream,
         string sessionId,
         Func<Task> initialize,
+        string? lastToken,
         ILogger logger,
         ChannelWriter<InternalBatchMessages<TValue>> channelWriter,
         IDeserializer<TValue> deserializer
     ) : base(
         stream,
         logger,
         sessionId,
-        initialize
+        initialize,
+        lastToken
     )
     {
         _readerConfig = config;

src/Ydb.Sdk/src/Services/Topic/TopicSession.cs

@@ -17,13 +17,14 @@ protected TopicSession(
         IBidirectionalStream<TFromClient, TFromServer> stream,
         ILogger logger,
         string sessionId,
-        Func<Task> initialize)
+        Func<Task> initialize,
+        string? lastToken)
     {
         Stream = stream;
         Logger = logger;
         SessionId = sessionId;
         _initialize = initialize;
-        _lastToken = stream.AuthToken;
+        _lastToken = lastToken;
     }
 
     public bool IsActive => Volatile.Read(ref _isActive) == 1;
@@ -44,7 +45,7 @@ protected async void ReconnectSession()
 
     protected async Task SendMessage(TFromClient fromClient)
     {
-        var curAuthToken = Stream.AuthToken;
+        var curAuthToken = await Stream.AuthToken;
 
         if (!string.Equals(_lastToken, curAuthToken) && curAuthToken != null)
         {

src/Ydb.Sdk/src/Services/Topic/Writer/Writer.cs

@@ -203,7 +203,8 @@ private async Task Initialize()
 
             _logger.LogInformation("Writer session initialization started. WriterConfig: {WriterConfig}", _config);
 
-            var stream = _driver.BidirectionalStreamCall(TopicService.StreamWriteMethod, _writerGrpcRequestSettings);
+            var stream =
+                await _driver.BidirectionalStreamCall(TopicService.StreamWriteMethod, _writerGrpcRequestSettings);
 
             var initRequest = new StreamWriteMessage.Types.InitRequest { Path = _config.TopicPath };
             if (_config.ProducerId != null)
@@ -301,6 +302,7 @@ private async Task Initialize()
                     lastSeqNo: lastSeqNo,
                     sessionId: initResponse.SessionId,
                     initialize: Initialize,
+                    await stream.AuthToken,
                     logger: _logger,
                     inFlightMessages: _inFlightMessages
                 );
@@ -450,13 +452,15 @@ public WriterSession(
         long lastSeqNo,
         string sessionId,
         Func<Task> initialize,
+        string? lastToken,
         ILogger logger,
         ConcurrentQueue<MessageSending> inFlightMessages
     ) : base(
         stream,
         logger,
         sessionId,
-        initialize
+        initialize,
+        lastToken
     )
     {
         _config = config;