2
\$\begingroup\$
using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Collections.Immutable;
using System.Linq;
using System.Threading;
using System.Threading.Tasks;

The MessageBroker Class

Contains thread-safe structures to hold mappings of message types to handlers. Dispatches messages to those handlers. Supports messages with and without replies.

    public class MessageBroker
    {
        public ConcurrentDictionary<(string name, Type message), ConcurrentBag<WeakReference<BrokeredMessageHandler>>> BrokeredMessageHandlers =
            new ConcurrentDictionary<(string name, Type message), ConcurrentBag<WeakReference<BrokeredMessageHandler>>>();

        public ConcurrentDictionary<(string name, Type message), ConcurrentBag<WeakReference<BrokeredMessageWithReplyHandler>>> BrokeredMessageWithReplyHandlers =
            new ConcurrentDictionary<(string name, Type message), ConcurrentBag<WeakReference<BrokeredMessageWithReplyHandler>>>();

        private static MessageBroker _instance;

        public static MessageBroker Instance => _instance ?? (_instance = new MessageBroker());

        public async Task Send(
            IBrokeredMessageBase message
            , Action<Guid> sent = null
            , Action<Guid> complete = null
            , CancellationToken cancellationToken = default)
        {
            var key = (message.MessageName, message.GetType());

            if (BrokeredMessageHandlers.ContainsKey(key))
            {
                var subscribers = BrokeredMessageHandlers[key].ToImmutableList();

                var tasks = new List<Task>();

                subscribers.ForEach(s => tasks.Add(Task.Factory.StartNew(() =>
                {
                    if (s.TryGetTarget(out var target) && !cancellationToken.IsCancellationRequested) target(message);
                }, cancellationToken)));

                if (!cancellationToken.IsCancellationRequested)
                {
                    sent?.Invoke(message.MessageUID);

                    await Task.Run(() => { Task.WaitAll(tasks.ToArray(), cancellationToken); }, cancellationToken);

                    if (!cancellationToken.IsCancellationRequested)
                    {
                        complete?.Invoke(message.MessageUID);
                    }
                }
            }
        }

        public async Task SendWithReply(
            BrokeredMessageWithReplyBase message
            , Action<Guid> sent = null
            , Action<Guid> complete = null
            , CancellationToken cancellationToken = default)
        {
            var key = (message.MessageName, message.GetType());

            if (BrokeredMessageWithReplyHandlers.ContainsKey(key))
            {
                var subscribers = BrokeredMessageWithReplyHandlers[key].ToImmutableList();

                var tasks = new List<Task>();

                subscribers.ForEach(s => tasks.Add(new Task(() =>
                {
                    if (s.TryGetTarget(out var target)) target(message);
                })));

                if (!cancellationToken.IsCancellationRequested)
                {
                    sent?.Invoke(message.MessageUID);

                    await Task.Run(() => { Task.WaitAll(tasks.ToArray(), cancellationToken); }, cancellationToken);

                    if (!cancellationToken.IsCancellationRequested)
                    {
                        complete?.Invoke(message.MessageUID);
                    }
                }
            }
        }

        public void AddHandler(string messageName, Type messageType, BrokeredMessageHandler brokeredMessageHandler)
        {
            var key = (messageName, messageType);

            BrokeredMessageHandlers.TryGetValue(key, out var bag);

            if (bag == null)
            {
                bag = new ConcurrentBag<WeakReference<BrokeredMessageHandler>>();
                BrokeredMessageHandlers.TryAdd(key, bag);
            }

            var reference = new WeakReference<BrokeredMessageHandler>(brokeredMessageHandler);
            if (!bag.Contains(reference))
            {
                bag.Add(reference);
            }
        }

        public void AddHandlerWithReply(string messageName, Type messageType, BrokeredMessageWithReplyHandler brokeredMessageWithReplyHandler)
        {
            var key = (messageName, messageType);

            BrokeredMessageWithReplyHandlers.TryGetValue(key, out var bag);

            if (bag == null)
            {
                bag = new ConcurrentBag<WeakReference<BrokeredMessageWithReplyHandler>>();
                BrokeredMessageWithReplyHandlers.TryAdd(key, bag);
            }

            var reference = new WeakReference<BrokeredMessageWithReplyHandler>(brokeredMessageWithReplyHandler);
            if (!bag.Contains(reference))
            {
                bag.Add(reference);
            }
        }
    }

BrokeredMessageHandler & BrokeredMessageWithReplyHandler

Delegates for dispatching the messages to.

    public delegate void BrokeredMessageHandler(IBrokeredMessageBase message);

    public delegate void BrokeredMessageWithReplyHandler(BrokeredMessageWithReplyBase message);

Messages (Without Reply)

  • IBrokeredMessageBase is the base interface for all Brokered Messages
  • BrokeredMessageBase abstract class providing basic implementation.
  • BrokeredMessage<TParameter> Generic Parameterized Brokered Message.
  • BrokeredMessage non-Generic Parameterized Brokered Message.
    public interface IBrokeredMessageBase
    {
        string MessageName { get; }
        Guid MessageUID { get; }
        DateTimeOffset TimestampUTC { get; }
    }

    public abstract class BrokeredMessageBase : IBrokeredMessageBase
    {
        public string MessageName { get; }
        public Guid MessageUID { get; }
        public DateTimeOffset TimestampUTC { get; } = DateTimeOffset.UtcNow;

        protected BrokeredMessageBase(string name, Guid uid = default)
        {
            MessageName = name;
            MessageUID = uid != default ? uid : Guid.NewGuid();
        }
    }

    public class BrokeredMessage<TParameter> : BrokeredMessageBase
    {
        public TParameter Parameter { get; }
        public BrokeredMessage(TParameter parameter, string name, Guid uid = default) : base(name, uid)
        {
            Parameter = parameter;
        }
    }

    public class BrokeredMessage : BrokeredMessageBase
    {
        public object Parameter { get; }
        public BrokeredMessage(object parameter, string name, Guid uid = default) : base(name, uid)
        {
            Parameter = parameter;
        }
    }

Messages (With Reply)

  • BrokeredMessageWithReplyBase abstract class providing basic implementation.
  • BrokeredMessageWithReply<TParameter> Generic Parameterized Brokered Message.
  • BrokeredMessageWithReply non-Generic Parameterized Brokered Message.
    public abstract class BrokeredMessageWithReplyBase : BrokeredMessageBase
    {
        public ConcurrentBag<object> Replies { get; }

        protected BrokeredMessageWithReplyBase(string name, Guid uid = default) : base(name, uid)
        {
            Replies = new ConcurrentBag<object>();
        }
    }

    public class BrokeredMessageWithReply<TParameter> : BrokeredMessageBase
    {
        public TParameter Parameter { get; }
        public BrokeredMessageWithReply(TParameter parameter, string name, Guid uid = default) : base(name, uid)
        {
            Parameter = parameter;
        }
    }

    public class BrokeredMessageWithReply : BrokeredMessageBase
    {
        public object Parameter { get; }
        public BrokeredMessageWithReply(object parameter, string name, Guid uid = default) : base(name, uid)
        {
            Parameter = parameter;
        }
    }
\$\endgroup\$
2
\$\begingroup\$

MessageBroker

  • Both public BrokeredMessageHandlers and BrokeredMessageWithReplyHandlers should be readonly because you don't want that someone sets them to null from outside of the class.
  • For both method names public async Task Send() and public async Task SendWithReply() the suffix Async should be appended. See: https://docs.microsoft.com/en-us/dotnet/csharp/async#important-info-and-advice
  • If you need to get a value of a Dictionary<TKey, TValue> you shouldn't us ContainsKey() together with the Item property getter but TryGetValue(), because by using ContainsKey() in combination with the Item getter you are doing the check if the key exists twice.
    From the refernce source

    public bool ContainsKey(TKey key)
    {
        if (key == null) throw new ArgumentNullException("key");
    
        TValue throwAwayValue;
        return TryGetValue(key, out throwAwayValue);
    }
    
    public TValue this[TKey key]
    {
        get
        {
            TValue value;
            if (!TryGetValue(key, out value))
            {
                throw new KeyNotFoundException();
            }
            return value;
        }
        set
        {
            if (key == null) throw new ArgumentNullException("key");
            TValue dummy;
            TryAddInternal(key, value, true, true, out dummy);
        }
    }
    
\$\endgroup\$
2
\$\begingroup\$

It is difficult to review something without understanding the scenario it is going to be used in. So, please regard this as just some random thoughts which may or may not be relevant:

  1. Waiting inside Task.Run is almost never a good idea. You consume a thread from a thread pool to do nothing. Try using an event loop if possible.
  2. In SendWithReply I don't see where the tasks are run. Also it looks like it copies a lot of code from Send
  3. AddHandler is not thread-safe. If two threads write to the same key, one might fail and never know about it
  4. Can't think of any use for the Sent callback, because by the time it is called the tasks might have already been finished
\$\endgroup\$
0
\$\begingroup\$

Here is the updated code based on feedback:

using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Collections.Immutable;
using System.Linq;
using System.Threading;
using System.Threading.Tasks;

namespace Boltron.UI.UWP.Commands
{
    public class MessageBroker
    {
        private readonly ConcurrentDictionary<(string name, Type message), ConcurrentBag<Action<IBrokeredMessage>>> _brokeredMessageHandlers =
            new ConcurrentDictionary<(string name, Type message), ConcurrentBag<Action<IBrokeredMessage>>>();

        private readonly ConcurrentDictionary<(string name, Type message), ConcurrentBag<Action<IBrokeredMessageWithReply>>> _brokeredMessageWithReplyHandlers =
            new ConcurrentDictionary<(string name, Type message), ConcurrentBag<Action<IBrokeredMessageWithReply>>>();

        public static MessageBroker Instance { get; } = new MessageBroker();

        public async Task SendAsync(
            IBrokeredMessage message
            , Action<Guid> sent = null
            , Action<Guid> complete = null
            , CancellationToken cancellationToken = default)
        {
            var key = (message.MessageName, message.GetType());

            if (_brokeredMessageHandlers.TryGetValue(key, out var bag))
            {
                var subscribers = bag.ToImmutableList();

                await SendMessageAsync(message, subscribers, sent, complete, cancellationToken);
            }
        }

        public async Task SendWithReplyAsync(
            IBrokeredMessageWithReply message
            , Action<Guid> sent = null
            , Action<Guid> complete = null
            , CancellationToken cancellationToken = default)
        {
            var key = (message.MessageName, message.GetType());

            if (_brokeredMessageWithReplyHandlers.TryGetValue(key, out var bag))
            {
                var subscribers = bag.Cast<Action<IBrokeredMessage>>().ToImmutableList();

                await SendMessageAsync(message, subscribers, sent, complete, cancellationToken);
            }
        }

        private async Task SendMessageAsync(
            IBrokeredMessage message
            , ImmutableList<Action<IBrokeredMessage>> subscribers
            , Action<Guid> sent = null
            , Action<Guid> complete = null
            , CancellationToken cancellationToken = default)
        {
            var tasks = new List<Task>();

            subscribers.ForEach(s => tasks.Add(Task.Factory.StartNew(() =>
            {
                if (s != null && !cancellationToken.IsCancellationRequested)
                {
                    s(message);
                }
            }, cancellationToken)));

            if (!cancellationToken.IsCancellationRequested)
            {
                sent?.Invoke(message.MessageUid);

                await Task.Run(
                    () => Task.WaitAll(tasks.ToArray(), cancellationToken)
                    , cancellationToken);

                if (!cancellationToken.IsCancellationRequested)
                {
                    complete?.Invoke(message.MessageUid);
                }
            }

        }

        public void AddHandler(string messageName
            , Type messageType
            , Action<IBrokeredMessage> brokeredMessageHandler)
        {
            var key = (messageName, messageType);

            _brokeredMessageHandlers.TryGetValue(key, out var bag);

            if (bag == null)
            {
                bag = new ConcurrentBag<Action<IBrokeredMessage>>();
                _brokeredMessageHandlers.TryAdd(key, bag);
            }

            if (!bag.Contains(brokeredMessageHandler))
            {
                bag.Add(brokeredMessageHandler);
            }
        }

        public void AddHandlerWithReply(string messageName
            , Type messageType
            , Action<IBrokeredMessageWithReply> brokeredMessageWithReplyHandler)
        {
            var key = (messageName, messageType);

            _brokeredMessageWithReplyHandlers.TryGetValue(key, out var bag);

            if (bag == null)
            {
                bag = new ConcurrentBag<Action<IBrokeredMessageWithReply>>();
                _brokeredMessageWithReplyHandlers.TryAdd(key, bag);
            }

            if (!bag.Contains(brokeredMessageWithReplyHandler))
            {
                bag.Add(brokeredMessageWithReplyHandler);
            }
        }
    }

    public interface IBrokeredMessage
    {
        string MessageName { get; }
        Guid MessageUid { get; }
        DateTimeOffset TimestampUtc { get; }
    }

    public abstract class BrokeredMessageBase : IBrokeredMessage
    {
        public string MessageName { get; }
        public Guid MessageUid { get; }
        public DateTimeOffset TimestampUtc { get; } = DateTimeOffset.UtcNow;

        protected BrokeredMessageBase(string name, Guid uid = default)
        {
            MessageName = name;
            MessageUid = uid != default ? uid : Guid.NewGuid();
        }
    }

    public class BrokeredMessage<TParameter> : BrokeredMessageBase
    {
        public TParameter Parameter { get; }
        public BrokeredMessage(TParameter parameter, string name, Guid uid = default) : base(name, uid)
        {
            Parameter = parameter;
        }
    }

    public class BrokeredMessage : BrokeredMessageBase
    {
        public object Parameter { get; }
        public BrokeredMessage(object parameter, string name, Guid uid = default) : base(name, uid)
        {
            Parameter = parameter;
        }
    }

    public interface IBrokeredMessageWithReply : IBrokeredMessage
    {

    }

    public abstract class BrokeredMessageWithReplyBase<TReturn> : BrokeredMessageBase, IBrokeredMessageWithReply
    {
        private readonly object _owner;
        private ConcurrentBag<TReturn> Replies { get; }

        protected BrokeredMessageWithReplyBase(string name, object owner, Guid uid = default) : base(name, uid)
        {
            _owner = owner;
            Replies = new ConcurrentBag<TReturn>();
        }

        public void AddReply(TReturn reply)
        {
            Replies.Add(reply);
        }

        public IEnumerable<TReturn> GetReplies(object claimant)
        {
            if (claimant != _owner) yield break;

            foreach (var item in Replies)
            {
                yield return item;
            }
        }
    }

    public class BrokeredMessageWithReply<TParameter, TReturn> : BrokeredMessageWithReplyBase<TReturn>
    {
        public TParameter Parameter { get; }
        public BrokeredMessageWithReply(TParameter parameter, string name, Guid uid = default) : base(name, uid)
        {
            Parameter = parameter;
        }
    }

    public class BrokeredMessageWithReply<TReturn> : BrokeredMessageWithReplyBase<TReturn>
    {
        public object Parameter { get; }
        public BrokeredMessageWithReply(object parameter, string name, Guid uid = default) : base(name, uid)
        {
            Parameter = parameter;
        }
    }
}
```
\$\endgroup\$

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Not the answer you're looking for? Browse other questions tagged or ask your own question.