entvex commented on code in PR #162:
URL: https://github.com/apache/pulsar-dotpulsar/pull/162#discussion_r1253110913
##########
src/DotPulsar/Internal/Consumer.cs:
##########
@@ -86,140 +133,191 @@ public async ValueTask DisposeAsync()
if (Interlocked.Exchange(ref _isDisposed, 1) != 0)
return;
- _eventRegister.Register(new ConsumerDisposed(_correlationId));
- await DisposeChannel().ConfigureAwait(false);
+ foreach (var subConsumer in _subConsumers.Values)
+ {
+ await subConsumer.DisposeAsync().ConfigureAwait(false);
+ }
}
- private async ValueTask DisposeChannel()
+ public async ValueTask<IMessage<TMessage>> Receive(CancellationToken
cancellationToken)
{
- await
_channel.ClosedByClient(CancellationToken.None).ConfigureAwait(false);
- await _channel.DisposeAsync().ConfigureAwait(false);
+ var iterations = 0;
+ while (_subConsumerIndex < _subConsumers.Count)
+ {
+ _subConsumerIndex++;
+ iterations++;
+ if (_subConsumerIndex == _subConsumers.Count)
+ _subConsumerIndex = 0;
+
+ var receiveTask =
_receiveTaskQueueForSubConsumers[_subConsumerIndex];
+ if (receiveTask == null)
+ {
+ var receiveTaskValueTask =
_subConsumers.ElementAt(_subConsumerIndex).Value.Receive(cancellationToken);
+
+ if (receiveTaskValueTask.IsCompleted)
+ {
+ return receiveTaskValueTask.Result;
+ }
+ {
+ _receiveTaskQueueForSubConsumers[_subConsumerIndex] =
receiveTaskValueTask.AsTask(); //.AsTask
+ }
+ }
+ else
+ {
+ if (receiveTask.IsCompleted)
+ {
+ _receiveTaskQueueForSubConsumers[_subConsumerIndex] = null;
+ return receiveTask.Result;
+ }
+ }
+ if (iterations == _subConsumers.Count)
+ await
Task.WhenAny(_receiveTaskQueueForSubConsumers).ConfigureAwait(false);
+ }
+ return null!; //TODO should never reach here. Should we throw an
exception or something ?
Review Comment:
While(true)
--
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: commits-unsubscr...@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
