feat: Init MVP Writer implementation (#206)

Author: KirillKurdyukov

src/Ydb.Sdk/src/Driver.cs

@@ -367,8 +367,7 @@ public async ValueTask<bool> MoveNextAsync()
         }
     }
 
-    internal sealed class BidirectionalStream<TRequest, TResponse> : IAsyncEnumerator<TResponse>,
-        IAsyncEnumerable<TResponse>
+    internal sealed class BidirectionalStream<TRequest, TResponse> : IDisposable
     {
         private readonly AsyncDuplexStreamingCall<TRequest, TResponse> _bidirectionalStream;
         private readonly Action _rpcErrorAction;
@@ -394,13 +393,6 @@ public async Task Write(TRequest request)
             }
         }
 
-        public ValueTask DisposeAsync()
-        {
-            _bidirectionalStream.Dispose();
-
-            return default;
-        }
-
         public async ValueTask<bool> MoveNextAsync()
         {
             try
@@ -417,9 +409,9 @@ public async ValueTask<bool> MoveNextAsync()
 
         public TResponse Current => _bidirectionalStream.ResponseStream.Current;
 
-        public IAsyncEnumerator<TResponse> GetAsyncEnumerator(CancellationToken cancellationToken = new())
+        public void Dispose()
         {
-            return this;
+            _bidirectionalStream.Dispose();
         }
     }
 

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

@@ -7,9 +7,9 @@ protected YdbTopicException(string message) : base(message)
     }
 }
 
-public class YdbProducerException : YdbTopicException
+public class YdbWriterException : YdbTopicException
 {
-    public YdbProducerException(string message) : base(message)
+    public YdbWriterException(string message) : base(message)
     {
     }
 }

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

@@ -0,0 +1,8 @@
+namespace Ydb.Sdk.Services.Topic;
+
+public interface IWriter<TValue>
+{
+    public Task<WriteResult> WriteAsync(TValue data);
+
+    public Task<WriteResult> WriteAsync(Message<TValue> message);
+}

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

@@ -0,0 +1,17 @@
+namespace Ydb.Sdk.Services.Topic;
+
+public class Message<TValue>
+{
+    public Message(TValue data)
+    {
+        Data = data;
+    }
+
+    public DateTime Timestamp { get; set; } = DateTime.Now;
+
+    public TValue Data { get; }
+
+    public List<Metadata> Metadata { get; } = new();
+}
+
+public record Metadata(string Key, byte[] Value);

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

@@ -0,0 +1,59 @@
+using Microsoft.Extensions.Logging;
+
+namespace Ydb.Sdk.Services.Topic;
+
+internal abstract class TopicSession<TFromClient, TFromServer> : IDisposable
+{
+    private readonly Func<Task> _initialize;
+
+    protected readonly Driver.BidirectionalStream<TFromClient, TFromServer> Stream;
+    protected readonly ILogger Logger;
+    protected readonly string SessionId;
+
+    private int _isActive = 1;
+    private bool _disposed;
+
+    protected TopicSession(Driver.BidirectionalStream<TFromClient, TFromServer> stream, ILogger logger,
+        string sessionId, Func<Task> initialize)
+    {
+        Stream = stream;
+        Logger = logger;
+        SessionId = sessionId;
+        _initialize = initialize;
+    }
+
+    protected async void ReconnectSession()
+    {
+        if (Interlocked.CompareExchange(ref _isActive, 0, 1) == 0)
+        {
+            Logger.LogWarning("Skipping reconnect. A reconnect session has already been initiated");
+
+            return;
+        }
+
+        Logger.LogInformation("WriterSession[{SessionId}] has been deactivated, starting to reconnect", SessionId);
+
+        while (!_disposed)
+        {
+            try
+            {
+                await _initialize();
+                break;
+            }
+            catch (Exception e)
+            {
+                Logger.LogError(e, "Unable to reconnect the session due to the following error");
+            }
+        }
+    }
+
+    public void Dispose()
+    {
+        lock (this)
+        {
+            _disposed = true;
+        }
+
+        Stream.Dispose();
+    }
+}

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

@@ -0,0 +1,40 @@
+using Ydb.Topic;
+
+namespace Ydb.Sdk.Services.Topic;
+
+public class WriteResult
+{
+    private readonly long _offset;
+
+    internal WriteResult(StreamWriteMessage.Types.WriteResponse.Types.WriteAck ack)
+    {
+        switch (ack.MessageWriteStatusCase)
+        {
+            case StreamWriteMessage.Types.WriteResponse.Types.WriteAck.MessageWriteStatusOneofCase.Written:
+                Status = PersistenceStatus.Written;
+                _offset = ack.Written.Offset;
+                break;
+            case StreamWriteMessage.Types.WriteResponse.Types.WriteAck.MessageWriteStatusOneofCase.Skipped:
+                Status = PersistenceStatus.AlreadyWritten;
+                break;
+            case StreamWriteMessage.Types.WriteResponse.Types.WriteAck.MessageWriteStatusOneofCase.None:
+            default:
+                throw new YdbWriterException($"Unexpected WriteAck status: {ack.MessageWriteStatusCase}");
+        }
+    }
+
+    public PersistenceStatus Status { get; }
+
+    public bool TryGetOffset(out long offset)
+    {
+        offset = _offset;
+
+        return Status == PersistenceStatus.Written;
+    }
+}
+
+public enum PersistenceStatus
+{
+    Written,
+    AlreadyWritten
+}