Skip to content

Commit 1897607

Browse files
KirillKurdyukovKirillKurdyukov
committed
next impl reader iteration
1 parent 7f0afee commit 1897607

File tree

3 files changed

+154
-34
lines changed

3 files changed

+154
-34
lines changed

‎src/Ydb.Sdk/src/Services/Topic/IReader.cs‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -6,5 +6,5 @@ public interface IReader<TValue> : IDisposable
66
{
77
public ValueTask<Message<TValue>> ReadAsync(CancellationToken cancellationToken = default);
88

9-
public ValueTask<IReadOnlyList<Message<TValue>>> ReadBatchAsync(CancellationToken cancellationToken = default);
9+
public ValueTask<BatchMessage<TValue>> ReadBatchAsync(CancellationToken cancellationToken = default);
1010
}

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

Lines changed: 47 additions & 11 deletions
Original file line numberDiff line numberDiff line change
@@ -1,17 +1,32 @@
1-
using Google.Protobuf;
2-
using Google.Protobuf.Collections;
3-
using Google.Protobuf.WellKnownTypes;
1+
using System.Collections.Immutable;
42
using Ydb.Topic;
53

64
namespace Ydb.Sdk.Services.Topic.Reader;
75

86
public class Message<TValue>
97
{
10-
internal Message(TValue data, string topic, string producerId)
8+
private readonly OffsetsRange _offsetsRange;
9+
private readonly ReaderSession _readerSession;
10+
11+
internal Message(
12+
TValue data,
13+
string topic,
14+
long partitionId,
15+
string producerId,
16+
DateTime createdAt,
17+
ImmutableArray<Metadata> metadata,
18+
OffsetsRange offsetsRange,
19+
ReaderSession readerSession)
1120
{
1221
Data = data;
1322
Topic = topic;
23+
PartitionId = partitionId;
1424
ProducerId = producerId;
25+
CreatedAt = createdAt;
26+
Metadata = metadata;
27+
28+
_offsetsRange = offsetsRange;
29+
_readerSession = readerSession;
1530
}
1631

1732
public TValue Data { get; }
@@ -21,25 +36,46 @@ internal Message(TValue data, string topic, string producerId)
2136
/// </summary>
2237
public string Topic { get; }
2338

39+
public long PartitionId { get; }
40+
2441
public string ProducerId { get; }
2542

26-
public Task Commit()
43+
public DateTime CreatedAt { get; }
44+
45+
public ImmutableArray<Metadata> Metadata { get; }
46+
47+
internal long Start => _offsetsRange.Start;
48+
internal long End => _offsetsRange.End;
49+
50+
public Task CommitAsync()
2751
{
28-
throw new NotImplementedException();
52+
return _readerSession.CommitOffsetRange(_offsetsRange);
2953
}
3054
}
3155

3256
public class BatchMessage<TValue>
3357
{
34-
public BatchMessage(IReadOnlyCollection<Message<TValue>> batch)
58+
private readonly ReaderSession _readerSession;
59+
60+
public ImmutableArray<Message<TValue>> Batch { get; }
61+
62+
internal BatchMessage(
63+
ImmutableArray<Message<TValue>> batch,
64+
ReaderSession readerSession)
3565
{
3666
Batch = batch;
67+
_readerSession = readerSession;
3768
}
3869

39-
public IReadOnlyCollection<Message<TValue>> Batch { get; }
40-
41-
public Task Commit()
70+
public Task CommitBatchAsync()
4271
{
43-
throw new NotImplementedException();
72+
if (Batch.Length == 0)
73+
{
74+
return Task.CompletedTask;
75+
}
76+
77+
var offsetsRange = new OffsetsRange { Start = Batch.First().Start, End = Batch.Last().End };
78+
79+
return _readerSession.CommitOffsetRange(offsetsRange);
4480
}
4581
}

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

Lines changed: 106 additions & 22 deletions
Original file line numberDiff line numberDiff line change
@@ -1,4 +1,5 @@
11
using System.Collections.Concurrent;
2+
using System.Collections.Immutable;
23
using System.Net.Sockets;
34
using System.Threading.Channels;
45
using Google.Protobuf;
@@ -42,14 +43,54 @@ internal Reader(IDriver driver, ReaderConfig config, IDeserializer<TValue> deser
4243
_ = Initialize();
4344
}
4445

45-
public ValueTask<Message<TValue>> ReadAsync(CancellationToken cancellationToken = default)
46+
public async ValueTask<Message<TValue>> ReadAsync(CancellationToken cancellationToken = default)
4647
{
47-
throw new NotImplementedException();
48+
while (await _receivedMessagesChannel.Reader.WaitToReadAsync(cancellationToken))
49+
{
50+
if (_receivedMessagesChannel.Reader.TryPeek(out var batchInternalMessage))
51+
{
52+
if (batchInternalMessage.InternalMessages.TryDequeue(out var message))
53+
{
54+
return message.ToPublicMessage(_deserializer, batchInternalMessage.ReaderSession);
55+
}
56+
57+
if (!_receivedMessagesChannel.Reader.TryRead(out _))
58+
{
59+
throw new ReaderException("Detect race condition on ReadAsync operation");
60+
}
61+
}
62+
else
63+
{
64+
throw new ReaderException("Detect race condition on ReadAsync operation");
65+
}
66+
}
67+
68+
throw new ReaderException("Reader is disposed");
4869
}
4970

50-
public ValueTask<IReadOnlyList<Message<TValue>>> ReadBatchAsync(CancellationToken cancellationToken = default)
71+
public async ValueTask<BatchMessage<TValue>> ReadBatchAsync(CancellationToken cancellationToken = default)
5172
{
52-
throw new NotImplementedException();
73+
while (await _receivedMessagesChannel.Reader.WaitToReadAsync(cancellationToken))
74+
{
75+
if (!_receivedMessagesChannel.Reader.TryRead(out var batchInternalMessage))
76+
{
77+
throw new ReaderException("Detect race condition on ReadBatchAsync operation");
78+
}
79+
80+
if (batchInternalMessage.InternalMessages.Count == 0)
81+
{
82+
continue;
83+
}
84+
85+
return new BatchMessage<TValue>(
86+
batchInternalMessage.InternalMessages
87+
.Select(message => message.ToPublicMessage(_deserializer, batchInternalMessage.ReaderSession))
88+
.ToImmutableArray(),
89+
batchInternalMessage.ReaderSession
90+
);
91+
}
92+
93+
throw new ReaderException("Reader is disposed");
5394
}
5495

5596
private async Task Initialize()
@@ -177,7 +218,7 @@ public void Dispose()
177218
try
178219
{
179220
_disposeCts.Cancel();
180-
221+
181222
_readerSession?.Dispose();
182223
}
183224
finally
@@ -212,6 +253,7 @@ internal class ReaderSession : TopicSession<MessageFromClient, MessageFromServer
212253
{
213254
private readonly ChannelWriter<InternalBatchMessage> _channelWriter;
214255
private readonly ConcurrentDictionary<long, PartitionSession> _partitionSessions = new();
256+
private readonly ConcurrentQueue<TaskCompletionSource> _tcsOnCommitedMessages = new();
215257

216258
private long _memoryUsageMaxBytes;
217259

@@ -258,6 +300,11 @@ public async void RunProcessingTopic()
258300
}
259301
}
260302

303+
public async Task CommitOffsetRange(OffsetsRange offsetsRange)
304+
{
305+
throw new NotImplementedException();
306+
}
307+
261308
private async Task HandleReadResponse()
262309
{
263310
var readResponse = Stream.Current.ReadResponse;
@@ -277,20 +324,25 @@ private async Task HandleReadResponse()
277324
{
278325
foreach (var messageData in batch.MessageData)
279326
{
280-
internalBatchMessages.Enqueue(
281-
new InternalMessage(
282-
messageData.Data,
283-
new OffsetsRange { Start = partitionSession.CommitedOffset, End = messageData.Offset },
284-
messageData.CreatedAt,
285-
messageData.MetadataItems
286-
));
327+
internalBatchMessages.Enqueue(new InternalMessage(
328+
data: messageData.Data,
329+
topic: partitionSession.TopicPath,
330+
partitionId: partitionSession.PartitionId,
331+
producerId: batch.ProducerId,
332+
offsetsRange: new OffsetsRange
333+
{ Start = partitionSession.CommitedOffset, End = messageData.Offset },
334+
createdAt: messageData.CreatedAt,
335+
metadataItems: messageData.MetadataItems
336+
));
287337

288338
partitionSession.CommitedOffset = endOffsetBatch = messageData.Offset + 1;
289339
}
290340
}
291341

292342
await _channelWriter.WriteAsync(new InternalBatchMessage(
293-
new OffsetsRange { Start = startOffsetBatch, End = endOffsetBatch }, internalBatchMessages)
343+
new OffsetsRange { Start = startOffsetBatch, End = endOffsetBatch },
344+
internalBatchMessages,
345+
this)
294346
);
295347
}
296348
else
@@ -348,34 +400,66 @@ internal class InternalMessage
348400
{
349401
public InternalMessage(
350402
ByteString data,
403+
string topic,
404+
long partitionId,
405+
string producerId,
351406
OffsetsRange offsetsRange,
352-
Timestamp createAt,
407+
Timestamp createdAt,
353408
RepeatedField<MetadataItem> metadataItems)
354409
{
355410
Data = data;
411+
Topic = topic;
412+
PartitionId = partitionId;
413+
ProducerId = producerId;
356414
OffsetsRange = offsetsRange;
357-
CreateAt = createAt;
415+
CreatedAt = createdAt;
358416
MetadataItems = metadataItems;
359417
}
360418

361-
public ByteString Data { get; }
419+
private ByteString Data { get; }
420+
421+
private string Topic { get; }
422+
423+
private long PartitionId { get; }
424+
425+
private string ProducerId { get; }
426+
427+
private OffsetsRange OffsetsRange { get; }
362428

363-
public OffsetsRange OffsetsRange { get; }
429+
private Timestamp CreatedAt { get; }
364430

365-
public Timestamp CreateAt { get; }
431+
private RepeatedField<MetadataItem> MetadataItems { get; }
366432

367-
public RepeatedField<MetadataItem> MetadataItems { get; }
433+
internal Message<TValue> ToPublicMessage<TValue>(IDeserializer<TValue> deserializer, ReaderSession readerSession)
434+
{
435+
return new Message<TValue>(
436+
data: deserializer.Deserialize(Data.ToByteArray()),
437+
topic: Topic,
438+
partitionId: PartitionId,
439+
producerId: ProducerId,
440+
createdAt: CreatedAt.ToDateTime(),
441+
metadata: MetadataItems.Select(item => new Metadata(item.Key, item.Value.ToByteArray())).ToImmutableArray(),
442+
offsetsRange: OffsetsRange,
443+
readerSession: readerSession
444+
);
445+
}
368446
}
369447

370448
internal class InternalBatchMessage
371449
{
372-
public InternalBatchMessage(OffsetsRange batchOffsetsRange, Queue<InternalMessage> internalMessages)
450+
public InternalBatchMessage(
451+
OffsetsRange batchOffsetsRange,
452+
Queue<InternalMessage> internalMessages,
453+
ReaderSession readerSession)
373454
{
374455
BatchOffsetsRange = batchOffsetsRange;
375456
InternalMessages = internalMessages;
457+
ReaderSession = readerSession;
376458
}
377459

378-
public OffsetsRange BatchOffsetsRange { get; }
460+
internal OffsetsRange BatchOffsetsRange { get; }
461+
462+
internal Queue<InternalMessage> InternalMessages { get; }
379463

380-
public Queue<InternalMessage> InternalMessages { get; }
464+
internal ReaderSession ReaderSession { get; }
381465
}

0 commit comments

Comments
 (0)