FCC/Applications/Share/dotnet_WayneLib_WayneLibraries/Common/MessageBus/ThreadedBus.cs

using System;
using System.Collections.Generic;
using System.Linq;
using System.Threading;

namespace Wayne.Lib.MessageBus
{
    /// <summary>
    /// A threaded bus implementation. The publish method
    /// will execute all subscribe actions.
    /// </summary>
    public class ThreadedBus : IBus
    {
        readonly object subscriptionsLock = new object();
        readonly List<SubscriptionBase> subscriptions = new List<SubscriptionBase>();

        public virtual void Publish(string topic, object message)
        {
            SubscriptionBase[] iterationSubscriptions;
            lock (subscriptionsLock)
            {
                iterationSubscriptions = subscriptions.ToArray();
            }

            iterationSubscriptions
                .Where(x => Any.Match(topic, x.TopicFilter))
                .Where(x => x.Type.IsInstanceOfType(message))
                .ForEach(x =>
                {
                    ThreadPool.QueueUserWorkItem(y => x.Invoke(topic, message));
                });
        }

        public virtual IDisposable Subscribe<T>(string topicFilter, Action<string, T> action)
        {
            var subscription = new Subscription<T>()
            {
                TopicFilter = topicFilter,
                InvokeAction = action
            };

            lock (subscriptionsLock)
            {
                subscriptions.Add(subscription);
            }

            return new ActionDisposable(() =>
            {
                lock (subscriptionsLock)
                {
                    subscriptions.Remove(subscription);
                }
            });
        }
    }
}