blankensteiner commented on code in PR #146:
URL: https://github.com/apache/pulsar-dotpulsar/pull/146#discussion_r1145897117
##########
samples/Consuming/Program.cs:
##########
@@ -52,8 +52,9 @@ private static async Task ConsumeMessages(IConsumer<string>
consumer, Cancellati
{
await foreach (var message in consumer.Messages(cancellationToken))
{
- Console.WriteLine($"Received: {message.Value()}");
- await consumer.Acknowledge(message, cancellationToken);
+ Console.WriteLine($"Received: {message?.Value()}");
Review Comment:
message is never null
##########
src/DotPulsar/DotPulsar.csproj:
##########
@@ -25,6 +25,7 @@
<PackageReference Include="HashDepot" Version="2.0.3" />
<PackageReference Include="Microsoft.Extensions.ObjectPool"
Version="7.0.4" />
<PackageReference Include="Microsoft.SourceLink.GitHub" Version="1.1.1"
PrivateAssets="All" />
+ <PackageReference Include="Nito.AsyncEx" Version="5.1.2" />
Review Comment:
Let's do this without onboarding a new dependency
##########
src/DotPulsar/Internal/Consumer.cs:
##########
@@ -86,140 +96,407 @@ public async ValueTask DisposeAsync()
if (Interlocked.Exchange(ref _isDisposed, 1) != 0)
return;
- _eventRegister.Register(new ConsumerDisposed(_correlationId));
- await DisposeChannel().ConfigureAwait(false);
+ _cts.Cancel();
+ _cts.Dispose();
+
+ _state.SetState(ConsumerState.Closed);
+
+ using (_lock.ReaderLock())
+ {
+ foreach (var consumer in _consumers.Values)
+ {
+ await consumer.DisposeAsync().ConfigureAwait(false);
+ }
+ }
}
- private async ValueTask DisposeChannel()
+ private async Task Setup()
{
- await
_channel.ClosedByClient(CancellationToken.None).ConfigureAwait(false);
- await _channel.DisposeAsync().ConfigureAwait(false);
- }
+ await Task.Yield();
- public async ValueTask<IMessage<TMessage>> Receive(CancellationToken
cancellationToken)
- => await _executor.Execute(() => InternalReceive(cancellationToken),
cancellationToken).ConfigureAwait(false);
+ try
+ {
+ await _executor.Execute(Monitor, _cts.Token).ConfigureAwait(false);
+ }
+ catch (Exception exception)
+ {
+ if (_cts.IsCancellationRequested)
+ return;
- public async ValueTask Acknowledge(MessageId messageId, CancellationToken
cancellationToken)
- => await InternalAcknowledge(messageId, CommandAck.AckType.Individual,
cancellationToken).ConfigureAwait(false);
+ _faultException = exception;
+ _state.SetState(ConsumerState.Faulted);
+ }
+ }
+ private async Task Monitor()
+ {
+ _numberOfPartitions = await GetNumberOfPartitions(Topic,
_cts.Token).ConfigureAwait(false);
+ _isPartitioned = _numberOfPartitions != 0;
+ var monitoringTasks = new List<Task<ConsumerStateChanged>>();
- public async ValueTask AcknowledgeCumulative(MessageId messageId,
CancellationToken cancellationToken)
- => await InternalAcknowledge(messageId, CommandAck.AckType.Cumulative,
cancellationToken).ConfigureAwait(false);
+ using (_lock.ReaderLock())
+ {
+ if (_isPartitioned)
+ {
+
+ for (var partition = 0; partition < _numberOfPartitions;
++partition)
+ {
+ var partitionedTopicName =
getPartitonedTopicName(partition);
+
+ var consumer = CreateSubConsumer(partitionedTopicName);
+ _ = _consumers.TryAdd(partitionedTopicName, consumer);
+
monitoringTasks.Add(consumer.StateChangedFrom(ConsumerState.Disconnected,
_cts.Token).AsTask());
+ }
+ }
+
+ else
+ {
+ var consumer = CreateSubConsumer(Topic);
+ _ = _consumers.TryAdd(Topic, consumer);
+
monitoringTasks.Add(consumer.StateChangedFrom(ConsumerState.Disconnected,
_cts.Token).AsTask());
+ }
+
+ Interlocked.Exchange(ref _consumerCount, monitoringTasks.Count);
+ }
+ var activeConsumers = 0;
+ while (true)
+ {
+ await Task.WhenAny(monitoringTasks).ConfigureAwait(false);
+
+ for (var i = 0; i < monitoringTasks.Count; ++i)
+ {
+ var task = monitoringTasks[i];
+ if (!task.IsCompleted)
+ continue;
+ var state = task.Result.ConsumerState;
+ switch (state)
+ {
+ case ConsumerState.Active:
+ ++activeConsumers;
+ break;
+ case ConsumerState.Disconnected:
+ --activeConsumers;
+ break;
+ case ConsumerState.ReachedEndOfTopic:
+ _state.SetState(ConsumerState.ReachedEndOfTopic);
+ return;
+ case ConsumerState.Faulted:
+ _state.SetState(ConsumerState.Faulted);
+ return;
+ case ConsumerState.Unsubscribed:
+ _state.SetState(ConsumerState.Unsubscribed);
+ return;
+ }
+
+ monitoringTasks[i] =
task.Result.Consumer.StateChangedFrom(state, _cts.Token).AsTask();
+ }
+
+ if (activeConsumers == 0)
+ _state.SetState(ConsumerState.Disconnected);
+ else if (activeConsumers == monitoringTasks.Count)
+ _state.SetState(ConsumerState.Active);
+ else
+ _state.SetState(ConsumerState.PartiallyActive);
+ }
+ }
- public async ValueTask
RedeliverUnacknowledgedMessages(IEnumerable<MessageId> messageIds,
CancellationToken cancellationToken)
+ private SubConsumer<TMessage> CreateSubConsumer(string topic)
{
- var command = new CommandRedeliverUnacknowledgedMessages();
- command.MessageIds.AddRange(messageIds.Select(messageId =>
messageId.ToMessageIdData()));
- await _executor.Execute(() =>
InternalRedeliverUnacknowledgedMessages(command, cancellationToken),
cancellationToken).ConfigureAwait(false);
- }
+ var correlationId = Guid.NewGuid();
+ var consumerName = _options.ConsumerName ??
$"Consumer-{correlationId:N}";
- public async ValueTask RedeliverUnacknowledgedMessages(CancellationToken
cancellationToken)
- => await
RedeliverUnacknowledgedMessages(Enumerable.Empty<MessageId>(),
cancellationToken).ConfigureAwait(false);
+ var subscribe = new CommandSubscribe
+ {
+ ConsumerName = consumerName,
+ InitialPosition = (CommandSubscribe.InitialPositionType)
_options.InitialPosition,
+ PriorityLevel = _options.PriorityLevel,
+ ReadCompacted = _options.ReadCompacted,
+ ReplicateSubscriptionState = _options.ReplicateSubscriptionState,
+ Subscription = _options.SubscriptionName,
+ Topic = topic,
+ Type = (CommandSubscribe.SubType) _options.SubscriptionType
+ };
+
+ foreach (var property in _options.SubscriptionProperties)
+ {
+ var keyValue = new KeyValue { Key = property.Key, Value =
property.Value };
+ subscribe.SubscriptionProperties.Add(keyValue);
+ }
- public async ValueTask Unsubscribe(CancellationToken cancellationToken)
+ var messagePrefetchCount = _options.MessagePrefetchCount;
+ var messageFactory = new MessageFactory<TMessage>(_options.Schema);
+ var batchHandler = new BatchHandler<TMessage>(true, messageFactory);
+ var decompressorFactories =
CompressionFactories.DecompressorFactories();
+
+ var factory = new ConsumerChannelFactory<TMessage>(correlationId,
_processManager, _connectionPool, subscribe, messagePrefetchCount,
batchHandler, messageFactory, decompressorFactories);
+ var stateManager = new
StateManager<ConsumerState>(ConsumerState.Disconnected, ConsumerState.Closed,
ConsumerState.ReachedEndOfTopic, ConsumerState.Faulted);
+ var initialChannel = new NotReadyChannel<TMessage>();
+ var executor = new Executor(correlationId, _processManager,
_exceptionHandler);
+ var consumer = new SubConsumer<TMessage>(correlationId, ServiceUrl,
_options.SubscriptionName, topic, _processManager, initialChannel, executor,
stateManager, factory);
+ var process = new ConsumerProcess(correlationId, stateManager,
consumer, _options.SubscriptionType == SubscriptionType.Failover);
+ _processManager.Add(process);
+ process.Start();
+ return consumer;
+ }
+
+ private async Task<uint> GetNumberOfPartitions(string topic,
CancellationToken cancellationToken)
{
- var unsubscribe = new CommandUnsubscribe();
- await _executor.Execute(() => InternalUnsubscribe(unsubscribe,
cancellationToken), cancellationToken).ConfigureAwait(false);
+ var connection = await _connectionPool.FindConnectionForTopic(topic,
cancellationToken).ConfigureAwait(false);
+ var commandPartitionedMetadata = new
PulsarApi.CommandPartitionedTopicMetadata { Topic = topic };
+ var response = await connection.Send(commandPartitionedMetadata,
cancellationToken).ConfigureAwait(false);
+
+
response.Expect(PulsarApi.BaseCommand.Type.PartitionedMetadataResponse);
+
+ if (response.PartitionMetadataResponse.Response ==
PulsarApi.CommandPartitionedTopicMetadataResponse.LookupType.Failed)
+ response.PartitionMetadataResponse.Throw();
+
+ return response.PartitionMetadataResponse.Partitions;
}
- public async ValueTask Seek(MessageId messageId, CancellationToken
cancellationToken)
+ private void ThrowIfDisposed()
+ {
+ if (_isDisposed != 0)
+ throw new ConsumerDisposedException(GetType().FullName!);
+ }
+ public async ValueTask Acknowledge(MessageId messageId, CancellationToken
cancellationToken = default)
{
- var seek = new CommandSeek { MessageId = messageId.ToMessageIdData() };
- await _executor.Execute(() => InternalSeek(seek, cancellationToken),
cancellationToken).ConfigureAwait(false);
+ ThrowIfDisposed();
+
+ var sourceTopic = Topic;
+ if (_isPartitioned)
+ {
+ sourceTopic = getPartitonedTopicName(messageId.Partition);
+ }
+ await _executor.Execute(() =>
+ {
+ ThrowIfNotActive();
+
+ using (_lock.ReaderLock())
+ {
+ return _consumers[sourceTopic].Acknowledge(messageId,
cancellationToken);
+ }
+ }, cancellationToken)
+ .ConfigureAwait(false);
}
- public async ValueTask Seek(ulong publishTime, CancellationToken
cancellationToken)
+ public async ValueTask AcknowledgeCumulative(MessageId messageId,
CancellationToken cancellationToken)
{
- var seek = new CommandSeek { MessagePublishTime = publishTime };
- await _executor.Execute(() => InternalSeek(seek, cancellationToken),
cancellationToken).ConfigureAwait(false);
+ ThrowIfDisposed();
+
+ await _executor.Execute(() =>
+ {
+ ThrowIfNotActive();
+
+ using (_lock.ReaderLock())
+ {
+ var sourceTopic = Topic;
+ if (_isPartitioned)
+ {
+ sourceTopic =
getPartitonedTopicName(messageId.Partition);
+ }
+ return
_consumers[sourceTopic].AcknowledgeCumulative(messageId, cancellationToken);
+ }
+ }, cancellationToken)
+ .ConfigureAwait(false);
}
public async ValueTask<MessageId> GetLastMessageId(CancellationToken
cancellationToken)
{
+ ThrowIfDisposed();
+
var getLastMessageId = new CommandGetLastMessageId();
- return await _executor.Execute(() =>
InternalGetLastMessageId(getLastMessageId, cancellationToken),
cancellationToken).ConfigureAwait(false);
+ return await _executor.Execute(() =>
GetLastMessageId(getLastMessageId, cancellationToken),
cancellationToken).ConfigureAwait(false);
}
- private void Guard()
+ private async ValueTask<MessageId>
GetLastMessageId(CommandGetLastMessageId command, CancellationToken
cancellationToken)
{
- if (_isDisposed != 0)
- throw new ConsumerDisposedException(GetType().FullName!);
+ ThrowIfNotActive();
- if (_faultException is not null)
- throw new ConsumerFaultedException(_faultException);
+ if (_isPartitioned)
+ {
+ throw new NotImplementedException("GetLastMessageId is not
implemented for partitioned topics");
+ }
+ using (_lock.ReaderLock())
+ {
+ return await
_consumers.First().Value.GetLastMessageId(cancellationToken).ConfigureAwait(false);
+ }
}
- public async Task EstablishNewChannel(CancellationToken cancellationToken)
+ public async ValueTask<IMessage<TMessage>> Receive(CancellationToken
cancellationToken = default)
{
- var channel = await _executor.Execute(() =>
_factory.Create(cancellationToken), cancellationToken).ConfigureAwait(false);
+ ThrowIfDisposed();
- var oldChannel = _channel;
- if (oldChannel is not null)
- await oldChannel.DisposeAsync().ConfigureAwait(false);
-
- _channel = channel;
+ return await _executor.Execute(() =>
ReceiveMessage(cancellationToken), cancellationToken).ConfigureAwait(false);
}
- public async ValueTask CloseChannel(CancellationToken cancellationToken)
- => await
_channel.ClosedByClient(cancellationToken).ConfigureAwait(false);
-
- public async ValueTask ChannelFaulted(Exception exception)
+ private async ValueTask<IMessage<TMessage>>
ReceiveMessage(CancellationToken cancellationToken)
{
- _faultException = exception;
- await DisposeChannel().ConfigureAwait(false);
+ ThrowIfNotActive();
+
+ using (_lock.ReaderLock())
+ {
+ if (_messagesQueue.TryDequeue(out var message))
+ {
+ return message;
+ }
+ var cts = new CancellationTokenSource();
+ var linkedCts =
CancellationTokenSource.CreateLinkedTokenSource(cancellationToken, cts.Token);
+ while (!cancellationToken.IsCancellationRequested)
+ {
+ var done = false;
+
+ Task<IMessage<TMessage>>[] receiveTasks =
_consumers.Values.Select(consumer =>
consumer.Receive(linkedCts.Token).AsTask()).ToArray();
+ await Task.WhenAny(receiveTasks).ConfigureAwait(false);
+
+ try
+ {
+ receiveTasks.Where(t => t.IsCompleted).ToList().ForEach(t
=>
+ {
+ if (t.Result == null)
+ {
+ return;
+ }
+
+ done = true;
+ _messagesQueue.Enqueue(t.Result);
+ }
+ );
+ }
+ catch (Exception exception)
+ {
+ if (linkedCts.IsCancellationRequested)
+ {
+ cts.Cancel();
+ }
+ else
+ {
+ throw exception;
Review Comment:
rethrow?
--
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]
