next iteration impl reader

Author: KirillKurdyukov

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

@@ -0,0 +1,85 @@
+using System.Collections.Immutable;
+using Google.Protobuf;
+using Google.Protobuf.Collections;
+using Google.Protobuf.WellKnownTypes;
+using Ydb.Topic;
+
+namespace Ydb.Sdk.Services.Topic.Reader;
+
+internal class InternalMessage
+{
+    public InternalMessage(
+        ByteString data,
+        string topic,
+        long partitionId,
+        string producerId,
+        OffsetsRange offsetsRange,
+        Timestamp createdAt,
+        RepeatedField<MetadataItem> metadataItems, 
+        int dataSize)
+    {
+        Data = data;
+        Topic = topic;
+        PartitionId = partitionId;
+        ProducerId = producerId;
+        OffsetsRange = offsetsRange;
+        CreatedAt = createdAt;
+        MetadataItems = metadataItems;
+        DataSize = dataSize;
+    }
+
+    private ByteString Data { get; }
+
+    private string Topic { get; }
+
+    private long PartitionId { get; }
+
+    private string ProducerId { get; }
+
+    private OffsetsRange OffsetsRange { get; }
+
+    private Timestamp CreatedAt { get; }
+
+    private RepeatedField<MetadataItem> MetadataItems { get; }
+    
+    private int DataSize { get; }
+
+    internal Message<TValue> ToPublicMessage<TValue>(IDeserializer<TValue> deserializer, ReaderSession readerSession)
+    {
+        return new Message<TValue>(
+            data: deserializer.Deserialize(Data.ToByteArray()),
+            topic: Topic,
+            partitionId: PartitionId,
+            producerId: ProducerId,
+            createdAt: CreatedAt.ToDateTime(),
+            metadata: MetadataItems.Select(item => new Metadata(item.Key, item.Value.ToByteArray())).ToImmutableArray(),
+            offsetsRange: OffsetsRange,
+            readerSession: readerSession
+        );
+    }
+}
+
+internal class InternalBatchMessage
+{
+    public InternalBatchMessage(
+        OffsetsRange batchOffsetsRange,
+        Queue<InternalMessage> internalMessages,
+        ReaderSession readerSession)
+    {
+        BatchOffsetsRange = batchOffsetsRange;
+        InternalMessages = internalMessages;
+        ReaderSession = readerSession;
+    }
+
+    internal OffsetsRange BatchOffsetsRange { get; }
+
+    internal Queue<InternalMessage> InternalMessages { get; }
+
+    internal ReaderSession ReaderSession { get; }
+}
+
+internal record CommitSending(
+    OffsetsRange OffsetsRange,
+    long PartitionSessionId,
+    TaskCompletionSource<TopicPartitionOffset> TcsTopicPartitionOffset
+);

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

@@ -49,7 +49,7 @@ internal Message(
 
     public Task CommitAsync()
     {
-        return _readerSession.CommitOffsetRange(_offsetsRange);
+        return _readerSession.CommitOffsetRange(_offsetsRange, PartitionId);
     }
 }
 
@@ -76,6 +76,19 @@ public Task CommitBatchAsync()
 
         var offsetsRange = new OffsetsRange { Start = Batch.First().Start, End = Batch.Last().End };
 
-        return _readerSession.CommitOffsetRange(offsetsRange);
+        return _readerSession.CommitOffsetRange(offsetsRange, Batch.First().PartitionId);
     }
 }
+
+public class TopicPartitionOffset
+{
+    public TopicPartitionOffset(long offset, long partitionId)
+    {
+        Offset = offset;
+        PartitionId = partitionId;
+    }
+
+    public long Offset { get; }
+
+    public long PartitionId { get; }
+}