Support for (optional) ack timeout and nack delay for consumers

src/DotPulsar/ConsumerOptions.cs

@@ -15,6 +15,7 @@
 namespace DotPulsar
 {
     using DotPulsar.Abstractions;
+    using System;
 
     /// <summary>
     /// The consumer building options.
@@ -101,5 +102,16 @@ public ConsumerOptions(string subscriptionName, string topic)
         /// Set the topic for this consumer. This is required.
         /// </summary>
         public string Topic { get; set; }
+
+        /// <summary>
+        /// Delay to wait before redelivering messages that failed to be processed.
+        /// When an application uses IConsumer.NegativeAcknowledge(Message), failed messages are redelivered after a fixed timeout.
+        /// </summary>
+        public TimeSpan NegativeAcknowledgementRedeliveryDelay { get; set; }
+
+        /// <summary>
+        /// Timeout of unacked messages
+        /// </summary>
+        public TimeSpan AcknowledgementTimeout { get; set; }
     }
 }

src/DotPulsar/Internal/Abstractions/IMessageQueue.cs

@@ -0,0 +1,25 @@
+/*
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+namespace DotPulsar.Internal.Abstractions
+{
+    using System;
+
+    public interface IMessageQueue : IDequeue<MessagePackage>, IDisposable
+    {
+        void Acknowledge(MessageId messageId);
+
+        void NegativeAcknowledge(MessageId messageId);
+    }
+}

src/DotPulsar/Internal/Abstractions/IUnackedMessageTracker.cs

@@ -0,0 +1,30 @@
+/*
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+namespace DotPulsar.Internal.Abstractions
+{
+    using DotPulsar.Abstractions;
+    using System.Threading.Tasks;
+    using System.Threading;
+    using System;
+
+    public interface IUnackedMessageTracker : IDisposable
+    {
+        void Add(MessageId messageId);
+
+        void Ack(MessageId messageId);
+
+        Task Start(IConsumer consumer, CancellationToken cancellationToken = default);
+    }
+}

src/DotPulsar/Internal/ConsumerChannel.cs

@@ -24,7 +24,7 @@ namespace DotPulsar.Internal
     public sealed class ConsumerChannel : IConsumerChannel, IReaderChannel
     {
         private readonly ulong _id;
-        private readonly AsyncQueue<MessagePackage> _queue;
+        private readonly MessageQueue _queue;
         private readonly IConnection _connection;
         private readonly BatchHandler _batchHandler;
         private readonly CommandFlow _cachedCommandFlow;
@@ -35,7 +35,7 @@ public sealed class ConsumerChannel : IConsumerChannel, IReaderChannel
         public ConsumerChannel(
             ulong id,
             uint messagePrefetchCount,
-            AsyncQueue<MessagePackage> queue,
+            MessageQueue queue,
             IConnection connection,
             BatchHandler batchHandler)
         {
@@ -108,6 +108,9 @@ public async Task Send(CommandAck command, CancellationToken cancellationToken)
             }
 
             command.ConsumerId = _id;
+
+            _queue.Acknowledge(new MessageId(messageId));
+
             await _connection.Send(command, cancellationToken).ConfigureAwait(false);
         }
 

src/DotPulsar/Internal/ConsumerChannelFactory.cs

@@ -28,20 +28,25 @@ public sealed class ConsumerChannelFactory : IConsumerChannelFactory
         private readonly IExecute _executor;
         private readonly CommandSubscribe _subscribe;
         private readonly uint _messagePrefetchCount;
+        private readonly int _ackTimeoutMillis;
+        private readonly int _negativeAckRedeliveryDelayMicros;
         private readonly BatchHandler _batchHandler;
+        private readonly IUnackedMessageTracker _tracker;
 
         public ConsumerChannelFactory(
             Guid correlationId,
             IRegisterEvent eventRegister,
             IConnectionPool connectionPool,
             IExecute executor,
-            ConsumerOptions options)
+            ConsumerOptions options,
+            IUnackedMessageTracker tracker)
         {
             _correlationId = correlationId;
             _eventRegister = eventRegister;
             _connectionPool = connectionPool;
             _executor = executor;
             _messagePrefetchCount = options.MessagePrefetchCount;
+            _tracker = tracker;
 
             _subscribe = new CommandSubscribe
             {
@@ -64,9 +69,10 @@ private async ValueTask<IConsumerChannel> GetChannel(CancellationToken cancellat
         {
             var connection = await _connectionPool.FindConnectionForTopic(_subscribe.Topic, cancellationToken).ConfigureAwait(false);
             var messageQueue = new AsyncQueue<MessagePackage>();
+            var consumerMessageQueue = new MessageQueue(messageQueue, _tracker);
             var channel = new Channel(_correlationId, _eventRegister, messageQueue);
             var response = await connection.Send(_subscribe, channel, cancellationToken).ConfigureAwait(false);
-            return new ConsumerChannel(response.ConsumerId, _messagePrefetchCount, messageQueue, connection, _batchHandler);
+            return new ConsumerChannel(response.ConsumerId, _messagePrefetchCount, consumerMessageQueue, connection, _batchHandler);
         }
     }
 }

src/DotPulsar/Internal/InactiveUnackedMessageTracker.cs

@@ -0,0 +1,42 @@
+/*
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+namespace DotPulsar.Internal
+{
+    using Abstractions;
+    using DotPulsar.Abstractions;
+    using System.Threading;
+    using System.Threading.Tasks;
+
+    public class InactiveUnackedMessageTracker : IUnackedMessageTracker
+    {
+
+        public void Ack(MessageId messageId)
+        {
+            return;
+        }
+
+        public void Add(MessageId messageId)
+        {
+            return;
+        }
+
+        public Task Start(IConsumer consumer, CancellationToken cancellationToken = default) => Task.CompletedTask;
+
+        public void Dispose()
+        {
+            return;
+        }
+    }
+}

src/DotPulsar/Internal/MessageQueue.cs

@@ -0,0 +1,53 @@
+/*
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+namespace DotPulsar.Internal
+{
+    using Abstractions;
+    using System;
+    using System.Threading;
+    using System.Threading.Tasks;
+
+    public sealed class MessageQueue : IMessageQueue, IDequeue<MessagePackage>, IDisposable
+    {
+        private readonly AsyncQueue<MessagePackage> _queue;
+        private readonly IUnackedMessageTracker _tracker;
+
+        public MessageQueue(AsyncQueue<MessagePackage> queue, IUnackedMessageTracker tracker)
+        {
+            _queue = queue;
+            _tracker = tracker;
+        }
+
+        public async ValueTask<MessagePackage> Dequeue(CancellationToken cancellationToken = default)
+        {
+            var message = await _queue.Dequeue(cancellationToken).ConfigureAwait(false);
+            _tracker.Add(new MessageId(message.MessageId));
+            return message;
+        }
+
+        public void Acknowledge(MessageId obj) => _tracker.Ack(obj);
+
+        public void NegativeAcknowledge(MessageId obj)
+        {
+            throw new NotImplementedException();
+        }
+
+        public void Dispose()
+        {
+            _queue.Dispose();
+            _tracker.Dispose();
+        }
+    }
+}

src/DotPulsar/Internal/ReaderChannelFactory.cs

@@ -63,9 +63,11 @@ private async ValueTask<IReaderChannel> GetChannel(CancellationToken cancellatio
         {
             var connection = await _connectionPool.FindConnectionForTopic(_subscribe.Topic, cancellationToken).ConfigureAwait(false);
             var messageQueue = new AsyncQueue<MessagePackage>();
+            var tracker = new InactiveUnackedMessageTracker(); // No tracker for reader since readers don't ack.
+            var consumerMessageQueue = new MessageQueue(messageQueue, tracker);
             var channel = new Channel(_correlationId, _eventRegister, messageQueue);
             var response = await connection.Send(_subscribe, channel, cancellationToken).ConfigureAwait(false);
-            return new ConsumerChannel(response.ConsumerId, _messagePrefetchCount, messageQueue, connection, _batchHandler);
+            return new ConsumerChannel(response.ConsumerId, _messagePrefetchCount, consumerMessageQueue, connection, _batchHandler);
         }
     }
 }

src/DotPulsar/Internal/UnackedMessageTracker.cs

@@ -0,0 +1,115 @@
+/*
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+namespace DotPulsar.Internal
+{
+    using Abstractions;
+    using DotPulsar.Abstractions;
+    using System;
+    using System.Collections.Concurrent;
+    using System.Diagnostics;
+    using System.Linq;
+    using System.Collections.Generic;
+    using System.Threading;
+    using System.Threading.Tasks;
+
+    public readonly struct AwaitingAck
+    {
+        public MessageId MessageId { get; }
+        public long Timestamp { get; }
+
+        public AwaitingAck(MessageId messageId)
+        {
+            MessageId = messageId;
+            Timestamp = Stopwatch.GetTimestamp();
+        }
+
+        public TimeSpan Elapsed => TimeSpan.FromTicks(
+            (long) ((Stopwatch.GetTimestamp() - Timestamp) / (double) Stopwatch.Frequency * TimeSpan.TicksPerSecond));
+    }
+
+    public sealed class UnackedMessageTracker : IUnackedMessageTracker
+    {
+        private readonly TimeSpan _ackTimeout;
+        private readonly TimeSpan _pollingTimeout;
+        private readonly ConcurrentQueue<AwaitingAck> _awaitingAcks;
+        private readonly List<MessageId> _acked;
+        private readonly CancellationTokenSource _cancellationTokenSource;
+
+        public UnackedMessageTracker(TimeSpan ackTimeout, TimeSpan pollingTimeout)
+        {
+            _ackTimeout = ackTimeout;
+            _pollingTimeout = pollingTimeout;
+            _awaitingAcks = new ConcurrentQueue<AwaitingAck>();
+            _acked = new List<MessageId>();
+            _cancellationTokenSource = new CancellationTokenSource();
+        }
+
+        public void Add(MessageId messageId)
+        {
+            _awaitingAcks.Enqueue(new AwaitingAck(messageId));
+        }
+
+        public void Ack(MessageId messageId)
+        {
+            // We only need to store the highest cumulative ack we see (if there is one)
+            // and the MessageIds not included by that cumulative ack.
+            _acked.Add(messageId);
+        }
+
+        public Task Start(IConsumer consumer, CancellationToken cancellationToken = default)
+        {
+            CancellationToken token =
+              CancellationTokenSource.CreateLinkedTokenSource(
+                  _cancellationTokenSource.Token, cancellationToken).Token;
+
+            return Task.Run(async () =>
+            {
+                while (!token.IsCancellationRequested)
+                {
+                    var messages = CheckUnackedMessages();
+
+                    if (messages.Count() > 0)
+                        await consumer.RedeliverUnacknowledgedMessages(messages, token);
+
+                    await Task.Delay(_pollingTimeout, token);
+                }
+            }, token);
+        }
+
+        private IEnumerable<MessageId> CheckUnackedMessages()
+        {
+            var result = new List<MessageId>();
+
+            while (_awaitingAcks.TryPeek(out AwaitingAck awaiting)
+                && awaiting.Elapsed > _ackTimeout)
+            {
+                if (_awaitingAcks.TryDequeue(out awaiting))
+                {
+                    if (!_acked.Contains(awaiting.MessageId))
+                        result.Add(awaiting.MessageId);
+                    else
+                        _acked.Remove(awaiting.MessageId);
+                }
+            }
+
+            return result;
+        }
+
+        public void Dispose()
+        {
+            this._cancellationTokenSource.Cancel();
+        }
+    }
+}

src/DotPulsar/PulsarClient.cs

@@ -74,11 +74,15 @@ public IConsumer CreateConsumer(ConsumerOptions options)
             ThrowIfDisposed();
             var correlationId = Guid.NewGuid();
             var executor = new Executor(correlationId, _processManager, _exceptionHandler);
-            var factory = new ConsumerChannelFactory(correlationId, _processManager, _connectionPool, executor, options);
             var stateManager = new StateManager<ConsumerState>(ConsumerState.Disconnected, ConsumerState.Closed, ConsumerState.ReachedEndOfTopic, ConsumerState.Faulted);
             var consumer = new Consumer(correlationId, options.Topic, _processManager, new NotReadyChannel(), executor, stateManager);
             if (options.StateChangedHandler is not null)
                 _ = StateMonitor.MonitorConsumer(consumer, options.StateChangedHandler);
+            IUnackedMessageTracker unackedTracker = options.AcknowledgementTimeout == default
+                ? new InactiveUnackedMessageTracker()
+                : new UnackedMessageTracker(options.AcknowledgementTimeout, TimeSpan.FromSeconds(1));
+            unackedTracker.Start(consumer);
+            var factory = new ConsumerChannelFactory(correlationId, _processManager, _connectionPool, executor, options, unackedTracker);
             var process = new ConsumerProcess(correlationId, stateManager, factory, consumer, options.SubscriptionType == SubscriptionType.Failover);
             _processManager.Add(process);
             process.Start();