What actual library can be used as a message broker inside a .NET application?

Question

Message Queues

Hey!

I decided to dive a little into game development on Unity. I noticed that the code in all the tutorials and tutorials is disgusting. The methods of each game object are made public, the state changes from 100,500 different places, terrible noodles and refactoring difficulties are everywhere. Apparently, it's accepted in indie, but I want to try to do it the right way. Namely, to have a bus/message broker inside the game application. Let everyone throw events there, and those who need to know something - subscribe and receive everything they need.

In principle, regular C# tools allow you to write your own system of messages. But I soberly assess my junior abilities: it will work, but slowly. Plus, I’ll spend xs how much time writing, and I’m rolling into Unity completely after another :)

Therefore, I would like to include a ready-made messaging library in the project, which will:

• very fast
  
• very efficient in terms of memory and processor (so as not to hang mobile devices)
  
• free (preferably open source)
  
• supported
  

I repeat, all this will work inside the application, without any external brokers.

Please advise what is currently being used for such purposes in the .NET world?

Answer 1

Decided to stretch.
Here is the tire:

MessageBus.cs

using System;
using System.Collections.Generic;
using System.Diagnostics;
using System.Threading.Tasks;

namespace SimpleMessageBus
{
    /// <summary>
    /// Рассылает события заинтересованным подписчикам
    /// в рамках одного приложения.
    /// </summary>
    /// <remarks>Используются WeakReference, чтобы не было утечек памяти
    /// на случай, если кто-то забудет отписаться. Поэтому не рекомендуется
    /// использовать лямбда-обработчики событий.</remarks>
    public static class MessageBus
    {
        private static readonly Dictionary<string, List<WeakReference<Action<string, object>>>> Subscribers =
            new Dictionary<string, List<WeakReference<Action<string, object>>>>();

        /// <summary>
        /// Оформляет подписку на событие
        /// </summary>
        /// <param name="topic">Тип события</param>
        /// <param name="handler">Подписчик</param>
        public static void Subscribe(string topic,
            Action<string, object> handler)
        {
            if (String.IsNullOrEmpty(topic))
                throw new ArgumentNullException(nameof(topic));

            if (handler == null)
                throw new ArgumentNullException(nameof(handler));

            List<WeakReference<Action<string, object>>> handlers;
            lock (Subscribers)
                if (!Subscribers.TryGetValue(topic, out handlers))
                    handlers = Subscribers[topic] = new List<WeakReference<Action<string, object>>>();

            lock (handlers)
                handlers.Add(new WeakReference<Action<string, object>>(handler));
        }

        /// <summary>
        /// Отменяет подписку на событие
        /// </summary>
        /// <param name="topic">Тип события</param>
        /// <param name="handler">Подписчик</param>
        public static void Unsubscribe(string topic,
            Action<string, object> handler)
        {
            if (String.IsNullOrEmpty(topic))
                throw new ArgumentNullException(nameof(topic));

            if (handler == null)
                throw new ArgumentNullException(nameof(handler));

            List<WeakReference<Action<string, object>>> list;
            lock (Subscribers)
                if (!Subscribers.TryGetValue(topic, out list))
                    return;

            lock (list)
            {
                var i = 0;
                while (i < list.Count)
                {
                    var reference = list[i];
                    if (!reference.TryGetTarget(out var target))
                        list.RemoveAt(i); // Заодно очищаем список от мертвых подписчиков
                    else if (target == handler)
                    {
                        list.RemoveAt(i);
                        return;
                    }
                }
            }
        }

        /// <summary>
        /// Оповещает подписчиков о наступлении события
        /// </summary>
        /// <param name="topic">Тип события</param>
        /// <param name="data">Данные события</param>
        public static void Publish(string topic, object data = null)
        {
            if (String.IsNullOrEmpty(topic))
                throw new ArgumentNullException(nameof(topic));

            List<WeakReference<Action<string, object>>> list;
            lock (Subscribers)
                if (!Subscribers.TryGetValue(topic, out list))
                    return;

            var handlers = new List<Action<string, object>>();
            lock (list)
            {
                var i = 0;
                while (i < list.Count)
                {
                    var reference = list[i];
                    if (!reference.TryGetTarget(out var target))
                        list.RemoveAt(i);
                    else
                    {
                        handlers.Add(target);
                        ++i;
                    }
                }
            }

            // В Unity вызывать обработчики событий, наверное,
            // нужно как-то так:
            UnityEngine.WSA.Application.InvokeOnAppThread(() =>
            // А в обычном консольном приложении так:
            // Task.Run(() =>
            {
                foreach (var handler in handlers)
                {
                    try
                    {
                        handler.Invoke(topic, data);
                    }
                    catch (Exception ex)
                    {
                        Debug.WriteLine(ex);
                    }
                }
            }, false);
        }
    }
}

Here is an example of usage:

using System.Diagnostics;
using System.Threading.Tasks;

namespace SimpleMessageBus
{
    public static class Program
    {
        public static async Task Main(string[] args)
        {
            Debug.WriteLine("Started.");

            MessageBus.Subscribe("Rain", RainHandler);
            MessageBus.Publish("Rain");

            await Task.Delay(100); // Ждем, пока событие поступит
            MessageBus.Unsubscribe("Rain", RainHandler);
            
            Debug.WriteLine("Finished.");
        }

        private static void RainHandler(string topic, object data)
        {
            Debug.WriteLine("Event: " + topic);
        }
    }
}

PS Free. Opensource.
Upd. Added a challenge UnityEngine.WSA.Application.InvokeOnAppThread()- after all, we have the target platform Unity.