entvex commented on code in PR #162:
URL: https://github.com/apache/pulsar-dotpulsar/pull/162#discussion_r1288271118
##########
src/DotPulsar/Internal/Reader.cs:
##########
@@ -72,86 +163,158 @@ public bool IsFinalState()
public bool IsFinalState(ReaderState state)
=> _state.IsFinalState(state);
+ [Obsolete("GetLastMessageId is obsolete. Please use GetLastMessageIds
instead.")]
public async ValueTask<MessageId> GetLastMessageId(CancellationToken
cancellationToken)
{
- var getLastMessageId = new CommandGetLastMessageId();
- return await _executor.Execute(() =>
InternalGetLastMessageId(getLastMessageId, cancellationToken),
cancellationToken).ConfigureAwait(false);
+ await Guard(cancellationToken).ConfigureAwait(false);
+
+ if (_isPartitioned)
+ {
+ throw new NotSupportedException("GetLastMessageId can't be used on
partitioned topics. Please use GetLastMessageIds");
+ }
+ else
+ {
+ return await
_subReaders[Topic].GetLastMessageId(cancellationToken).ConfigureAwait(false);
+ }
}
- private async ValueTask<MessageId>
InternalGetLastMessageId(CommandGetLastMessageId command, CancellationToken
cancellationToken)
+ public async ValueTask<IEnumerable<MessageId>>
GetLastMessageIds(CancellationToken cancellationToken)
{
- Guard();
- return await _channel.Send(command,
cancellationToken).ConfigureAwait(false);
+ await Guard(cancellationToken).ConfigureAwait(false);
+
+ if (_isPartitioned)
+ {
+ Task<MessageId>[] getLastMessageIdsTasks = new
Task<MessageId>[_numberOfPartitions];
+
+ for (var i = 0; i < _subReaders.Count; i++)
+ {
+ var getLastMessageIdTask =
_subReaders.Values.ElementAt(i).GetLastMessageId(cancellationToken);
+ getLastMessageIdsTasks[i] = getLastMessageIdTask.AsTask();
+ }
+
+ //await all of the tasks.
+ await Task.WhenAll(getLastMessageIdsTasks).ConfigureAwait(false);
+
+ //collect MessageIds
+ List<MessageId> messageIds = new List<MessageId>();
+ for (var i = 0; i < _subReaders.Count; i++)
+ {
+ messageIds.Add(getLastMessageIdsTasks[i].Result);
+ }
+ return messageIds;
+ }
+ else
+ {
+ MessageId[] messageIds = new MessageId[1];
Review Comment:
yes!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
