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

src/DotPulsar/ConsumerOptions.cs

@@ -101,5 +101,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 int NegativeAckRedeliveryDelayMicros { get; set; }
+
+        /// <summary>
+        /// Timeout of unacked messages
+        /// </summary>
+        public int AckTimeoutMillis { get; set; }
     }
 }

src/DotPulsar/Internal/Abstractions/IMessageAcksTracker.cs

@@ -0,0 +1,13 @@
+namespace DotPulsar.Internal.Abstractions
+{
+    using DotPulsar.Abstractions;
+    using System.Threading;
+    using System.Threading.Tasks;
+    public interface IMessageAcksTracker<T>
+    {
+        T Add(T message);
+        T Ack(T message);
+        T Nack(T message);
+        Task StartTracker(IConsumer consumer, CancellationToken cancellationToken);
+    }
+}

src/DotPulsar/Internal/Abstractions/IMessageQueue.cs

@@ -0,0 +1,10 @@
+namespace DotPulsar.Internal.Abstractions
+{
+    using System;
+
+    public interface IMessageQueue : IDequeue<MessagePackage>, IDisposable
+    {
+        void Acknowledge(MessageId obj);
+        void NegativeAcknowledge(MessageId obj);
+    }
+}

src/DotPulsar/Internal/Abstractions/IUnackedMessageTracker.cs

@@ -0,0 +1,16 @@
+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/InactiveMessageAcksTracker.cs

@@ -0,0 +1,30 @@
+namespace DotPulsar.Internal
+{
+    using Abstractions;
+    using DotPulsar.Abstractions;
+    using DotPulsar.Exceptions;
+    using Events;
+    using Microsoft.Extensions.ObjectPool;
+    using PulsarApi;
+    using System;
+    using System.Collections.Generic;
+    using System.Linq;
+    using System.Runtime.CompilerServices;
+    using System.Threading;
+    using System.Threading.Tasks;
+    using System.Diagnostics;
+
+    public sealed class InactiveMessageAcksTracker : IMessageAcksTracker<MessageId>
+    {
+        public InactiveMessageAcksTracker() { }
+
+        public async Task StartTracker(IConsumer consumer, CancellationToken cancellationToken)
+        {
+            await Task.Yield();
+        }
+
+        public MessageId Add(MessageId message) => message;
+        public MessageId Ack(MessageId message) => message;
+        public MessageId Nack(MessageId message) => message;
+    }
+}

src/DotPulsar/Internal/InactiveUnackedMessageTracker.cs

@@ -0,0 +1,31 @@
+namespace DotPulsar.Internal
+{
+    using System.Threading;
+    using System.Threading.Tasks;
+    using Abstractions;
+    using DotPulsar.Abstractions;
+
+    public class InactiveUnackedMessageTracker : IUnackedMessageTracker
+    {
+        public InactiveUnackedMessageTracker()
+        {
+        }
+
+        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/MessageAcksTracker.cs

@@ -0,0 +1,103 @@
+namespace DotPulsar.Internal
+{
+    using Abstractions;
+    using DotPulsar.Abstractions;
+    using DotPulsar.Exceptions;
+    using Events;
+    using Microsoft.Extensions.ObjectPool;
+    using PulsarApi;
+    using System;
+    using System.Collections.Generic;
+    using System.Linq;
+    using System.Runtime.CompilerServices;
+    using System.Threading;
+    using System.Threading.Tasks;
+    using System.Diagnostics;
+
+    internal class Tracker
+    {
+        private readonly Stopwatch _timer;
+        private int maxTimeoutMs;
+
+        public Tracker(int timeoutMs)
+        {
+            maxTimeoutMs = timeoutMs;
+            _timer = new Stopwatch();
+            _timer.Start();
+        }
+
+        public bool IsTimedOut() => _timer.ElapsedMilliseconds > maxTimeoutMs;
+
+        public long msTillTimeout => maxTimeoutMs - _timer.ElapsedMilliseconds;
+
+        public void Reset(int newTimeoutMs)
+        {
+            maxTimeoutMs = newTimeoutMs;
+            _timer.Restart();
+        }
+    }
+
+    // TODO add mechnism to stop tracker when disposed
+    public sealed class MessageAcksTracker : IMessageAcksTracker<MessageId>
+    {
+        private readonly Dictionary<MessageId, Tracker> _trackers;
+        private readonly int _unackedTimeoutMs;
+        private readonly int _nackTimeoutMs;
+        private readonly int _trackerDelayMs;
+        public MessageAcksTracker(int unackedTimeoutMs, int nackTimeoutMs, int trackerDelayMs)
+        {
+            _unackedTimeoutMs = unackedTimeoutMs;
+            _nackTimeoutMs = nackTimeoutMs;
+            _trackerDelayMs = trackerDelayMs;
+            _trackers = new Dictionary<MessageId, Tracker>();
+        }
+
+        public async Task StartTracker(IConsumer consumer, CancellationToken cancellationToken)
+        {
+            await Task.Yield();
+
+            while (true)
+            {
+                await Task.Delay(_trackerDelayMs);
+
+                var messageIds = new List<MessageId>();
+                foreach (KeyValuePair<MessageId, Tracker> p in _trackers)
+                {
+                    if (p.Value.IsTimedOut())
+                        messageIds.Add(p.Key);
+                }
+
+                if (messageIds.Count() > 0)
+                    await consumer.RedeliverUnacknowledgedMessages(messageIds, cancellationToken).ConfigureAwait(false);
+
+            }
+        }
+        public MessageId Add(MessageId message)
+        {
+            if (!_trackers.ContainsKey(message))
+            {
+                _trackers.Add(message, new Tracker(_unackedTimeoutMs));
+            }
+
+            return message;
+        }
+        public MessageId Ack(MessageId message)
+        {
+            if (_trackers.ContainsKey(message))
+                _trackers.Remove(message);
+            return message;
+        }
+        public MessageId Nack(MessageId message)
+        {
+            if (_trackers.ContainsKey(message))
+            {
+                var timer = _trackers[message];
+                if (timer.msTillTimeout > _nackTimeoutMs)
+                    timer.Reset(_nackTimeoutMs);
+            }
+            else
+                _trackers.Add(message, new Tracker(_nackTimeoutMs));
+            return message;
+        }
+    }
+}

src/DotPulsar/Internal/MessageQueue.cs

@@ -0,0 +1,45 @@
+namespace DotPulsar.Internal
+{
+    using Abstractions;
+    using DotPulsar.Abstractions;
+    using DotPulsar.Exceptions;
+    using Events;
+    using Microsoft.Extensions.ObjectPool;
+    using PulsarApi;
+    using System;
+    using System.Collections.Generic;
+    using System.Linq;
+    using System.Runtime.CompilerServices;
+    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);
         }
     }
 }