Add AcknowledgeChannel.

Author: tosh-coding

src/AsyncFiberWorks/Channels/AcknowledgementChannel.cs

@@ -0,0 +1,41 @@
+using System;
+using System.Threading.Tasks;
+
+namespace AsyncFiberWorks.Channels
+{
+    /// <summary>
+    /// This channel is a publishing interface controlled by an acknowledgment.
+    /// </summary>
+    /// <typeparam name="TMessage"></typeparam>
+    /// <typeparam name="TAck"></typeparam>
+    public class AcknowledgementChannel<TMessage, TAck> : IAcknowledgementChannel<TMessage, TAck>
+    {
+        private readonly AcknowledgementMessageHandlerList<TMessage, TAck> _channel = new AcknowledgementMessageHandlerList<TMessage, TAck>();
+
+        /// <summary>
+        /// Subscribe a channel.
+        /// </summary>
+        /// <param name="messageReceiver">Subscriber.</param>
+        /// <returns>Unsubscriber.</returns>
+        public IDisposable Subscribe(IAcknowledgeMessageReceiver<TMessage, TAck> messageReceiver)
+        {
+            return this._channel.AddHandler(messageReceiver.OnReceive);
+        }
+
+        /// <summary>
+        /// Publish a message to subscribers.
+        /// </summary>
+        /// <param name="msg">The message to send.</param>
+        /// <param name="control">Publishing controller.</param>
+        /// <returns>Waiting for the publishing process to complete.</returns>
+        public async Task Publish(TMessage msg, IAcknowledgementControl<TMessage, TAck> control)
+        {
+            await _channel.Publish(msg, control);
+        }
+
+        ///<summary>
+        /// Number of subscribers
+        ///</summary>
+        public int NumSubscribers { get { return _channel.Count; } }
+    }
+}

src/AsyncFiberWorks/Channels/AcknowledgementMessageHandlerList.cs

@@ -0,0 +1,67 @@
+using System;
+using System.Collections.Generic;
+using System.Linq;
+using System.Threading.Tasks;
+
+namespace AsyncFiberWorks.Channels
+{
+    /// <summary>
+    /// List of message handlers with acknowledgement.
+    /// </summary>
+    /// <typeparam name="TMessage"></typeparam>
+    /// <typeparam name="TAck"></typeparam>
+    internal sealed class AcknowledgementMessageHandlerList<TMessage, TAck>
+    {
+        private object _lock = new object();
+        private LinkedList<Func<TMessage, Task<TAck>>> _handlers = new LinkedList<Func<TMessage, Task<TAck>>>();
+
+        /// <summary>
+        /// Add a message handler.
+        /// </summary>
+        /// <param name="action">A message handler.</param>
+        /// <returns>Function for removing the handler.</returns>
+        public IDisposable AddHandler(Func<TMessage, Task<TAck>> action)
+        {
+            _handlers.AddLast(action);
+
+            var unsubscriber = new Unsubscriber(() => {
+                lock (_lock)
+                {
+                    this._handlers.Remove(action);
+                }
+            });
+
+            return unsubscriber;
+        }
+
+        /// <summary>
+        /// Send a message to receive handlers.
+        /// </summary>
+        /// <param name="msg">The message to send.</param>
+        /// <param name="control">Publishing controller.</param>
+        /// <returns>A task that waits for IAcknowledgeControl.OnPublish to complete.</returns>
+        public async Task Publish(TMessage msg, IAcknowledgementControl<TMessage, TAck> control)
+        {
+            Func<TMessage, Task<TAck>>[] copied;
+            lock (_lock)
+            {
+                copied = _handlers.ToArray();
+            }
+            await control.OnPublish(msg, copied).ConfigureAwait(false);
+        }
+
+        /// <summary>
+        /// Number of handlers.
+        /// </summary>
+        public int Count
+        {
+            get
+            {
+                lock (_lock)
+                {
+                    return _handlers.Count;
+                }
+            }
+        }
+    }
+}

src/AsyncFiberWorks/Channels/DefaultAcknowledgementChannelSubscription.cs

@@ -0,0 +1,40 @@
+using System;
+using System.Threading.Tasks;
+
+namespace AsyncFiberWorks.Channels
+{
+    /// <summary>
+    /// Subscription for actions on a acknowledgement channel.
+    /// </summary>
+    /// <typeparam name="TMessage"></typeparam>
+    public class DefaultAcknowledgementChannelSubscription<TMessage> : IAcknowledgeMessageReceiver<TMessage, bool>
+    {
+        private readonly Func<TMessage, Task<bool>> _receiver;
+        private readonly IMessageFilter<TMessage> _filter;
+
+        /// <summary>
+        /// Construct the subscription
+        /// </summary>
+        /// <param name="receiver"></param>
+        /// <param name="filter"></param>
+        public DefaultAcknowledgementChannelSubscription(Func<TMessage, Task<bool>> receiver, IMessageFilter<TMessage> filter = null)
+        {
+            _receiver = receiver;
+            _filter = filter;
+        }
+
+        /// <summary>
+        /// Message receive handler with acknowledgement.
+        /// </summary>
+        /// <param name="msg">A received message.</param>
+        /// <returns>True if accepted, false if ignored.</returns>
+        public async Task<bool> OnReceive(TMessage msg)
+        {
+            if (_filter?.PassesProducerThreadFilter(msg) ?? true)
+            {
+                return await _receiver(msg);
+            }
+            return false;
+        }
+    }
+}

src/AsyncFiberWorks/Channels/DefaultAcknowledgementControl.cs

@@ -0,0 +1,35 @@
+using System;
+using System.Threading.Tasks;
+
+namespace AsyncFiberWorks.Channels
+{
+    /// <summary>
+    /// Call all handlers in the order in which they are registered.
+    /// Wait for the calls to complete one at a time before proceeding.
+    /// If ACK is false, it moves on to the next subscriber. If true, publish will be terminated at that point.
+    /// </summary>
+    /// <typeparam name="TMessage"></typeparam>
+    public class DefaultAcknowledgementControl<TMessage> : IAcknowledgementControl<TMessage, bool>
+    {
+        /// <summary>
+        /// Publish a message and accept an ACK.
+        /// </summary>
+        /// <param name="msg">The message to send.</param>
+        /// <param name="handlers">A list of message recipients.</param>
+        /// <returns>Wait for the publishing process to complete.</returns>
+        public async Task OnPublish(TMessage msg, Func<TMessage, Task<bool>>[] handlers)
+        {
+            if (handlers != null)
+            {
+                foreach (var h in handlers)
+                {
+                    bool ack = await h(msg).ConfigureAwait(false);
+                    if (ack)
+                    {
+                        break;
+                    }
+                }
+            }
+        }
+    }
+}

src/AsyncFiberWorks/Channels/IAcknowledgeMessageReceiver.cs

@@ -0,0 +1,19 @@
+using System.Threading.Tasks;
+
+namespace AsyncFiberWorks.Channels
+{
+    /// <summary>
+    /// Message receive handler with acknowledgement.
+    /// </summary>
+    /// <typeparam name="TMessage"></typeparam>
+    /// <typeparam name="TAck"></typeparam>
+    public interface IAcknowledgeMessageReceiver<TMessage, TAck>
+    {
+        /// <summary>
+        /// Message receive handler with acknowledgement.
+        /// </summary>
+        /// <param name="msg">A received message.</param>
+        /// <returns>Tasks waiting for subscribers to be received.</returns>
+        Task<TAck> OnReceive(TMessage msg);
+    }
+}

src/AsyncFiberWorks/Channels/IAcknowledgementChannel.cs

@@ -0,0 +1,13 @@
+using System;
+
+namespace AsyncFiberWorks.Channels
+{
+    /// <summary>
+    /// This channel is a publishing interface controlled by an acknowledgment.
+    /// </summary>
+    /// <typeparam name="TMessage"></typeparam>
+    /// <typeparam name="TAck"></typeparam>
+    public interface IAcknowledgementChannel<TMessage, TAck> : IAcknowledgementSubscriber<TMessage, TAck>, IAcknowledgementPublisher<TMessage, TAck>
+    {
+    }
+}

src/AsyncFiberWorks/Channels/IAcknowledgementControl.cs

@@ -0,0 +1,21 @@
+using System;
+using System.Threading.Tasks;
+
+namespace AsyncFiberWorks.Channels
+{
+    /// <summary>
+    /// Control message transmission.
+    /// </summary>
+    /// <typeparam name="TMessage"></typeparam>
+    /// <typeparam name="TAck"></typeparam>
+    public interface IAcknowledgementControl<TMessage, TAck>
+    {
+        /// <summary>
+        /// Handle the response from the notification destination.
+        /// </summary>
+        /// <param name="msg">The message to send.</param>
+        /// <param name="handlers">A list of message recipients.</param>
+        /// <returns>Wait for the publishing process to complete.</returns>
+        Task OnPublish(TMessage msg, Func<TMessage, Task<TAck>>[] handlers);
+    }
+}

src/AsyncFiberWorks/Channels/IAcknowledgementPublisher.cs

@@ -0,0 +1,20 @@
+using System.Threading.Tasks;
+
+namespace AsyncFiberWorks.Channels
+{
+    /// <summary>
+    /// Sender interface to channels with acknowledgments.
+    /// </summary>
+    /// <typeparam name="TMessage"></typeparam>
+    /// <typeparam name="TAck"></typeparam>
+    public interface IAcknowledgementPublisher<TMessage, TAck>
+    {
+        /// <summary>
+        /// Publish a message to subscribers.
+        /// </summary>
+        /// <param name="msg">The message to send.</param>
+        /// <param name="control">Publishing controller.</param>
+        /// <returns>Waiting for the publishing process to complete.</returns>
+        Task Publish(TMessage msg, IAcknowledgementControl<TMessage, TAck> control);
+    }
+}

src/AsyncFiberWorks/Channels/IAcknowledgementSubscriber.cs

@@ -0,0 +1,19 @@
+using System;
+
+namespace AsyncFiberWorks.Channels
+{
+    /// <summary>
+    /// Acknowledgement channel subscription interface.
+    /// </summary>
+    /// <typeparam name="TMessage"></typeparam>
+    /// <typeparam name="TAck"></typeparam>
+    public interface IAcknowledgementSubscriber<TMessage, TAck>
+    {
+        /// <summary>
+        /// Subscribe a channel.
+        /// </summary>
+        /// <param name="messageReceiver">Subscriber.</param>
+        /// <returns>Unsubscriber.</returns>
+        IDisposable Subscribe(IAcknowledgeMessageReceiver<TMessage, TAck> messageReceiver);
+    }
+}

src/AsyncFiberWorks/Channels/ReverseOrderAcknowledgementControl.cs

@@ -0,0 +1,36 @@
+using System;
+using System.Linq;
+using System.Threading.Tasks;
+
+namespace AsyncFiberWorks.Channels
+{
+    /// <summary>
+    /// Call the handlers in the reverse order in which they were registered.
+    /// Wait for the calls to complete one at a time before proceeding.
+    /// If ACK is false, it moves on to the next subscriber. If true, publish will be terminated at that point.
+    /// </summary>
+    /// <typeparam name="TMessage"></typeparam>
+    public class ReverseOrderAcknowledgementControl<TMessage> : IAcknowledgementControl<TMessage, bool>
+    {
+        /// <summary>
+        /// Publish a message and accept an ACK.
+        /// </summary>
+        /// <param name="msg">The message to send.</param>
+        /// <param name="handlers">A list of message recipients.</param>
+        /// <returns>Wait for the publishing process to complete.</returns>
+        public async Task OnPublish(TMessage msg, Func<TMessage, Task<bool>>[] handlers)
+        {
+            if (handlers != null)
+            {
+                foreach (var h in handlers.Reverse())
+                {
+                    bool ack = await h(msg).ConfigureAwait(false);
+                    if (ack)
+                    {
+                        break;
+                    }
+                }
+            }
+        }
+    }
+}