feat: updating Serializer / Deserializer, added stress Integration Test

Author: KirillKurdyukov

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

@@ -55,28 +55,15 @@ private class Int64Deserializer : IDeserializer<long>
     {
         public long Deserialize(byte[] data)
         {
-            if (data.Length != 8)
-            {
-                throw new ArgumentException(
-                    $"Deserializer<Long> encountered data of length ${data.Length}. Expecting data length to be 8");
-            }
-
-            return ((long)data[0] << 56) | ((long)data[1] << 48) | ((long)data[2] << 40) | ((long)data[3] << 32) |
-                   ((long)data[4] << 24) | ((long)data[5] << 16) | ((long)data[6] << 8) | data[7];
+            return BitConverter.ToInt64(data);
         }
     }
 
     private class Int32Deserializer : IDeserializer<int>
     {
         public int Deserialize(byte[] data)
         {
-            if (data.Length != 4)
-            {
-                throw new ArgumentException(
-                    $"Deserializer<Int32> encountered data of length ${data.Length}. Expecting data length to be 4");
-            }
-
-            return (data[0] << 24) | (data[1] << 16) | (data[2] << 8) | data[3];
+            return BitConverter.ToInt32(data);
         }
     }
 

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

@@ -49,31 +49,15 @@ 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
-            };
+            return BitConverter.GetBytes(data);
         }
     }
 
     private class Int32Serializer : ISerializer<int>
     {
         public byte[] Serialize(int data)
         {
-            return new[]
-            {
-                (byte)(data >> 24),
-                (byte)(data >> 16),
-                (byte)(data >> 8),
-                (byte)data
-            };
+            return BitConverter.GetBytes(data);
         }
     }
 

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

@@ -0,0 +1,27 @@
+using Xunit;
+using Ydb.Sdk.Services.Topic;
+
+namespace Ydb.Sdk.Tests.Topic;
+
+public class SerializerDeserializerUnitTests
+{
+    [Fact]
+    public void SerializeDeserialize_WhenSerializer64Deserializer32_ReturnInt32()
+    {
+        Assert.Equal(32, Deserializers.Int32.Deserialize(Serializers.Int32.Serialize(32)));
+    }
+
+    [Fact]
+    public void SerializeDeserialize_WhenSerializer64Deserializer64_ReturnInt64()
+    {
+        Assert.Equal(32 * 1_000_000_000L,
+            Deserializers.Int64.Deserialize(Serializers.Int64.Serialize(32 * 1_000_000_000L)));
+    }
+
+    [Fact]
+    public void SerializeDeserialize_WhenSerializerUtf8DeserializerUtf8_ReturnString()
+    {
+        Assert.Equal("abacaba",
+            Deserializers.Utf8.Deserialize(Serializers.Utf8.Serialize("abacaba")));
+    }
+}

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

@@ -1,7 +1,10 @@
+using Google.Protobuf.WellKnownTypes;
 using Xunit;
 using Ydb.Sdk.Services.Topic;
 using Ydb.Sdk.Services.Topic.Writer;
 using Ydb.Sdk.Tests.Fixture;
+using Ydb.Topic;
+using Consumer = Ydb.Sdk.Services.Topic.Consumer;
 
 namespace Ydb.Sdk.Tests.Topic;
 
@@ -44,4 +47,82 @@ public async Task WriteAsync_WhenTopicNotFound_ReturnNotFoundException()
         Assert.Equal(StatusCode.SchemeError, (await Assert.ThrowsAsync<WriterException>(
             () => writer.WriteAsync("hello world"))).Status.StatusCode);
     }
+
+    [Fact]
+    public async Task WriteAsync_When1000Messages_ReturnWriteResultIsPersisted()
+    {
+        const int messageCount = 1000;
+        var topicName = _topicName + "_stress";
+        var topicClient = new TopicClient(_driver);
+        var topicSettings = new CreateTopicSettings { Path = topicName };
+        topicSettings.Consumers.Add(new Consumer("Consumer"));
+        await topicClient.CreateTopic(topicSettings);
+
+        using var writer = new WriterBuilder<int>(_driver, topicName) { ProducerId = "producerId" }.Build();
+
+        var tasks = new List<Task>();
+        for (var i = 0; i < messageCount; i++)
+        {
+            var i1 = i;
+            tasks.Add(Task.Run(async () =>
+            {
+                // ReSharper disable once AccessToDisposedClosure
+                var message = await writer.WriteAsync(i1);
+
+                Assert.Equal(PersistenceStatus.Written, message.Status);
+            }));
+        }
+
+        await Task.WhenAll(tasks);
+
+        var initStream =
+            _driver.BidirectionalStreamCall(Ydb.Topic.V1.TopicService.StreamReadMethod, new GrpcRequestSettings());
+        await initStream.Write(new StreamReadMessage.Types.FromClient
+        {
+            InitRequest = new StreamReadMessage.Types.InitRequest
+            {
+                Consumer = "Consumer", ReaderName = "reader-test", TopicsReadSettings =
+                {
+                    new StreamReadMessage.Types.InitRequest.Types.TopicReadSettings
+                        { ReadFrom = new Timestamp(), Path = topicName }
+                }
+            }
+        });
+
+        var ans = 0;
+
+        await initStream.MoveNextAsync();
+        await initStream.Write(new StreamReadMessage.Types.FromClient
+        {
+            ReadRequest = new StreamReadMessage.Types.ReadRequest
+            {
+                BytesSize = 2000 * messageCount * sizeof(int)
+            }
+        });
+        await initStream.MoveNextAsync();
+        var startRequest = initStream.Current.StartPartitionSessionRequest;
+        await initStream.Write(new StreamReadMessage.Types.FromClient
+        {
+            StartPartitionSessionResponse = new StreamReadMessage.Types.StartPartitionSessionResponse
+            {
+                CommitOffset = startRequest.CommittedOffset,
+                PartitionSessionId = startRequest.PartitionSession.PartitionSessionId,
+            }
+        });
+        var receivedMessageCount = 0;
+        while (receivedMessageCount < messageCount)
+        {
+            await initStream.MoveNextAsync();
+            ans += initStream.Current.ReadResponse.PartitionData.Sum(data => data.Batches.Sum(batch =>
+                batch.MessageData.Sum(message =>
+                {
+                    receivedMessageCount++;
+                    return Deserializers.Int32.Deserialize(message.Data.ToByteArray());
+                })));
+        }
+
+        Assert.Equal(messageCount * (messageCount - 1) / 2, ans);
+
+        await topicClient.DropTopic(new DropTopicSettings { Path = topicName });
+    }
 }

src/Ydb.Sdk/tests/Topic/WriterMockTests.cs renamed to src/Ydb.Sdk/tests/Topic/WriterUnitTests.cs

@@ -13,13 +13,13 @@ namespace Ydb.Sdk.Tests.Topic;
 using WriterStream = IBidirectionalStream<StreamWriteMessage.Types.FromClient, StreamWriteMessage.Types.FromServer>;
 using FromClient = StreamWriteMessage.Types.FromClient;
 
-public class WriterMockTests
+public class WriterUnitTests
 {
     private readonly ITestOutputHelper _testOutputHelper;
     private readonly Mock<IDriver> _mockIDriver = new();
     private readonly Mock<WriterStream> _mockStream = new();
 
-    public WriterMockTests(ITestOutputHelper testOutputHelper)
+    public WriterUnitTests(ITestOutputHelper testOutputHelper)
     {
         _testOutputHelper = testOutputHelper;
         _mockIDriver.Setup(driver => driver.BidirectionalStreamCall(