commit

Author: KirillKurdyukov

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

@@ -0,0 +1,15 @@
+namespace Ydb.Sdk.Services.Topic;
+
+public class YdbTopicException : Exception
+{
+    protected YdbTopicException(string message) : base(message)
+    {
+    }
+}
+
+public class YdbProducerException : YdbTopicException
+{
+    public YdbProducerException(string message) : base(message)
+    {
+    }
+}

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

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

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

@@ -0,0 +1,59 @@
+using Ydb.Topic;
+using Ydb.Topic.V1;
+
+namespace Ydb.Sdk.Services.Topic;
+
+public class ProducerBuilder<TValue>
+{
+    private readonly ProducerConfig _config;
+
+    public ProducerBuilder(ProducerConfig config)
+    {
+        _config = config;
+    }
+
+    public ISerializer<TValue>? Serializer { get; set; }
+
+    public async Task<IProducer<TValue>> Build()
+    {
+        var stream = _config.Driver.BidirectionalStreamCall(TopicService.StreamWriteMethod,
+            GrpcRequestSettings.DefaultInstance);
+
+        var initRequest = new StreamWriteMessage.Types.InitRequest { Path = _config.TopicPath };
+        if (_config.ProducerId != null)
+        {
+            initRequest.ProducerId = _config.ProducerId;
+        }
+
+        if (_config.MessageGroupId != null)
+        {
+            initRequest.MessageGroupId = _config.MessageGroupId;
+        }
+
+        await stream.Write(new StreamWriteMessage.Types.FromClient { InitRequest = initRequest });
+        if (!await stream.MoveNextAsync())
+        {
+            throw new YdbProducerException("Write stream is closed by YDB server");
+        }
+
+        var receivedInitMessage = stream.Current;
+
+        Status.FromProto(receivedInitMessage.Status, receivedInitMessage.Issues).EnsureSuccess();
+
+        var initResponse = receivedInitMessage.InitResponse;
+
+        if (!initResponse.SupportedCodecs.Codecs.Contains((int)_config.Codec))
+        {
+            throw new YdbProducerException($"Topic is not supported codec: {_config.Codec}");
+        }
+
+        return new Producer<TValue>(
+            _config, initResponse, stream,
+            Serializer ?? (ISerializer<TValue>)(
+                Serializers.DefaultSerializers.TryGetValue(typeof(TValue), out var serializer)
+                    ? serializer
+                    : throw new YdbProducerException("The serializer is not set")
+            )
+        );
+    }
+}

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

@@ -0,0 +1,16 @@
+namespace Ydb.Sdk.Services.Topic;
+
+public class ProducerConfig
+{
+    public ProducerConfig(Driver driver, string topicPath)
+    {
+        Driver = driver;
+        TopicPath = topicPath;
+    }
+
+    public Driver Driver { get; }
+    public string TopicPath { get; }
+    public string? ProducerId { get; set; }
+    public string? MessageGroupId { get; set; }
+    public Codec Codec { get; set; } = Codec.Raw; // TODO Supported only Raw
+}

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

@@ -0,0 +1,85 @@
+using System.Text;
+
+namespace Ydb.Sdk.Services.Topic;
+
+public interface ISerializer<in TValue>
+{
+    public byte[] Serialize(TValue data);
+}
+
+public static class Serializers
+{
+    /// <summary>String (UTF8) serializer.</summary>
+    public static readonly ISerializer<string> Utf8 = new Utf8Serializer();
+
+    /// <summary>
+    /// System.Int64 (big endian, network byte order) serializer.
+    /// </summary>
+    public static readonly ISerializer<long> Int64 = new Int64Serializer();
+
+    /// <summary>
+    /// System.Int32 (big endian, network byte order) serializer.
+    /// </summary>
+    public static readonly ISerializer<int> Int32 = new Int32Serializer();
+
+    /// <summary>
+    /// System.Byte[] (nullable) serializer.</summary>
+    /// <remarks>Byte order is original order.</remarks>
+    public static readonly ISerializer<byte[]> ByteArray = new ByteArraySerializer();
+
+    internal static readonly Dictionary<System.Type, object> DefaultSerializers = new()
+    {
+        { typeof(int), Int32 },
+        { typeof(long), Int64 },
+        { typeof(string), Utf8 },
+        { typeof(byte[]), ByteArray }
+    };
+
+    private class Utf8Serializer : ISerializer<string>
+    {
+        public byte[] Serialize(string data)
+        {
+            return Encoding.UTF8.GetBytes(data);
+        }
+    }
+
+    private class Int64Serializer : ISerializer<long>
+    {
+        public byte[] Serialize(long data)
+        {
+            return new[]
+            {
+                (byte)(data >> 56),
+                (byte)(data >> 48),
+                (byte)(data >> 40),
+                (byte)(data >> 32),
+                (byte)(data >> 24),
+                (byte)(data >> 16),
+                (byte)(data >> 8),
+                (byte)data
+            };
+        }
+    }
+
+    private class Int32Serializer : ISerializer<int>
+    {
+        public byte[] Serialize(int data)
+        {
+            return new[]
+            {
+                (byte)(data >> 24),
+                (byte)(data >> 16),
+                (byte)(data >> 8),
+                (byte)data
+            };
+        }
+    }
+
+    private class ByteArraySerializer : ISerializer<byte[]>
+    {
+        public byte[] Serialize(byte[] data)
+        {
+            return data;
+        }
+    }
+}