micro fixex

Author: KirillKurdyukov

src/Ydb.Sdk/src/IDriver.cs

@@ -6,21 +6,21 @@ namespace Ydb.Sdk;
 
 public interface IDriver : IAsyncDisposable, IDisposable
 {
-    internal Task<TResponse> UnaryCall<TRequest, TResponse>(
+    public Task<TResponse> UnaryCall<TRequest, TResponse>(
         Method<TRequest, TResponse> method,
         TRequest request,
         GrpcRequestSettings settings)
         where TRequest : class
         where TResponse : class;
 
-    internal ServerStream<TResponse> ServerStreamCall<TRequest, TResponse>(
+    public ServerStream<TResponse> ServerStreamCall<TRequest, TResponse>(
         Method<TRequest, TResponse> method,
         TRequest request,
         GrpcRequestSettings settings)
         where TRequest : class
         where TResponse : class;
 
-    internal BidirectionalStream<TRequest, TResponse> BidirectionalStreamCall<TRequest, TResponse>(
+    public BidirectionalStream<TRequest, TResponse> BidirectionalStreamCall<TRequest, TResponse>(
         Method<TRequest, TResponse> method,
         GrpcRequestSettings settings)
         where TRequest : class
@@ -213,11 +213,16 @@ public async ValueTask<bool> MoveNextAsync()
     }
 }
 
-public sealed class BidirectionalStream<TRequest, TResponse> : IDisposable
+public class BidirectionalStream<TRequest, TResponse> : IDisposable
 {
     private readonly AsyncDuplexStreamingCall<TRequest, TResponse> _stream;
     private readonly Action<RpcException> _rpcErrorAction;
 
+    public BidirectionalStream()
+    {
+        
+    }
+    
     internal BidirectionalStream(
         AsyncDuplexStreamingCall<TRequest, TResponse> stream,
         Action<RpcException> rpcErrorAction)

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

@@ -1,5 +1,4 @@
 using System.Collections.Concurrent;
-using System.Transactions;
 using Google.Protobuf;
 using Google.Protobuf.WellKnownTypes;
 using Microsoft.Extensions.Logging;
@@ -303,17 +302,14 @@ public async Task Write(ConcurrentQueue<MessageSending> toSendBuffer)
             Volatile.Write(ref _seqNum, currentSeqNum);
             await Stream.Write(new MessageFromClient { WriteRequest = writeMessage });
         }
-        catch (TransactionException e)
+        catch (Driver.TransportException e)
         {
             Logger.LogError(e, "WriterSession[{SessionId}] have error on Write, last SeqNo={SeqNo}",
                 SessionId, Volatile.Read(ref _seqNum));
 
-            ReconnectSession();
+            ClearInFlightMessages(e);
 
-            while (_inFlightMessages.TryDequeue(out var sendData))
-            {
-                sendData.TaskCompletionSource.SetException(e);
-            }
+            ReconnectSession();
         }
     }
 
@@ -377,13 +373,23 @@ Completing task on exception...
                 }
             }
         }
-        catch (Exception e)
+        catch (Driver.TransportException e)
         {
             Logger.LogError(e, "WriterSession[{SessionId}] have error on processing writeAck", SessionId);
+
+            ClearInFlightMessages(e);
         }
         finally
         {
             ReconnectSession();
         }
     }
+
+    private void ClearInFlightMessages(Driver.TransportException e)
+    {
+        while (_inFlightMessages.TryDequeue(out var sendData))
+        {
+            sendData.TaskCompletionSource.SetException(e);
+        }
+    }
 }

src/Ydb.Sdk/tests/Topic/WriterIntegrationTests.cs

@@ -40,8 +40,7 @@ public async Task WriteAsync_WhenOneMessage_ReturnWritten()
     [Fact]
     public async Task WriteAsync_WhenTopicNotFound_ReturnNotFoundException()
     {
-        using var writer = new WriterBuilder<string>(_driver,
-                new WriterConfig(_topicName + "_not_found") { ProducerId = "producerId" })
+        using var writer = new WriterBuilder<string>(_driver, new WriterConfig(_topicName + "_not_found") { ProducerId = "producerId" })
             .Build();
 
         Assert.Equal(StatusCode.SchemeError, (await Assert.ThrowsAsync<WriterException>(

src/Ydb.Sdk/tests/Topic/WriterMockTests.cs

@@ -0,0 +1,43 @@
+using Grpc.Core;
+using Moq;
+using Xunit;
+using Ydb.Sdk.Services.Topic;
+using Ydb.Sdk.Services.Topic.Writer;
+using Ydb.Topic;
+
+namespace Ydb.Sdk.Tests.Topic;
+
+using WriterStream = BidirectionalStream<StreamWriteMessage.Types.FromClient, StreamWriteMessage.Types.FromServer>;
+
+public class WriterMockTests
+{
+    private readonly Mock<IDriver> _mockIDriver = new();
+    private readonly Mock<WriterStream> _mockStream = new();
+
+    public WriterMockTests()
+    {
+        _mockIDriver.Setup(driver => driver.BidirectionalStreamCall(
+            It.IsAny<Method<StreamWriteMessage.Types.FromClient, StreamWriteMessage.Types.FromServer>>(),
+            It.IsAny<GrpcRequestSettings>())
+        ).Returns(_mockStream.Object);
+    }
+
+    // [Fact]
+    public async Task NotStarted_Failed_Test()
+    {
+        var moveNextTry = new TaskCompletionSource<bool>();
+
+        _mockStream
+            .Setup(stream => stream.Write(It.IsAny<StreamWriteMessage.Types.FromClient>()))
+            .Returns(Task.CompletedTask);
+
+        _mockStream.SetupSequence(stream => stream.MoveNextAsync())
+            .ReturnsAsync(false)
+            .Returns(new ValueTask<bool>(moveNextTry.Task)); // For retry
+
+        using var writer = new WriterBuilder<int>(_mockIDriver.Object, new WriterConfig("/topic")
+            { ProducerId = "producerId" }).Build();
+
+        await Assert.ThrowsAsync<WriterException>(() => writer.WriteAsync(100));
+    }
+}