[Wzorce projektowe] Event Aggregator

Analogia z życia: 


Nowe wydania gazet pojawiają się przeważnie każdego dnia, co tydzień lub np. co miesiąc. Kiedy tylko wydawnictwo przygotuje całe wydanie, wysyła je do kiosku. Czytelnik zainteresowany swoimi ulubionymi gazetami udaje się do kiosku by nabyć nowe egzemplarze. Informacje na temat tego, czy nowe wydanie już jest dostępne uzyskuje właśnie od kioskarza.

Zastosowanie:

Event Aggregator upraszcza zarządzanie zdarzeniami przez wprowadzenie jednego scentralizowanego obiektu zarządzającego nimi. Dzięki temu powiązania między nadawcami i odbiorcami eventów są zredukowane do minimum. Zatem wzorca tego należy użyć wszędzie tam, gdzie mamy do czynienia z wieloma modułami, które muszą się ze sobą komunikować, a także w każdym innym przypadku, gdzie relacja między nadawcami a odbiorcami zdarzeń przedstawia się jako wiele do wielu. Dodatkowo zmniejszone zostaje ryzyko wycieków pamięci związanych z eventami, a wprowadzenie nowego zdarzenia nie zaburza struktury całości.

Zasada działania:

W omawianym wzorcu występują trzy rodzaje obiektów : nadawcy, odbiorcy i agregator. Zarówno nadawcy, jak i odbiorcy przechowują referencję do agregatora. Gdy zajdzie taka potrzeba nadawcy wywołują odpowiednie metody z agregatora notyfikując odbiorców. Odbiorcy wołają metody subskrypcji by odbierać wiadomości, które ich interesują. Odbiorca może na przykład implementować generyczny interfejs, mówiący o tym, które eventy chce subskrybować. Event Aggregator przechowuje jedynie słabe referencje (takie, które nie powstrzymają GC przed sprzątnięciem obiektu) do subskrybentów, co zapobiega wyciekom pamięci.

Przykład implementacyjny:

 public interface IEventAggregator
    {
        void Subscribe(object subscriber);
        void Publish<TEvent>(TEvent eventToSend);
    }

    public class SimpleEventAggregator : IEventAggregator
    {
        private readonly Dictionary<Type, List<WeakReference>> _subscribersList = 
            new Dictionary<Type, List<WeakReference>>();

        private readonly object _syncObject = new object();

        public void Subscribe(object subscriber)
        {
            lock (_syncObject)
            {
                var subscriberTypes = subscriber.GetType().GetInterfaces().Where(i => i.IsGenericType
                                                                                      &&
                                                                                      i.GetGenericTypeDefinition() ==
                                                                                      typeof (ISubscriber<>));
                var weakReference = new WeakReference(subscriber);
                foreach (Type subscriberType in subscriberTypes)
                {
                    var subscribers = GetSubscribers(subscriberType);
                    subscribers.Add(weakReference);
                }
            }
        }

        private List<WeakReference> GetSubscribers(Type subscriberType)
        {
            List<WeakReference> subscribers;
            lock (_syncObject)
            {
                var found = _subscribersList.TryGetValue(subscriberType, out subscribers);
                if(!found)
                {
                    subscribers = new List<WeakReference>();
                    _subscribersList.Add(subscriberType, subscribers);
                }
            }
            return subscribers;
        }

        public void Publish<TEvent>(TEvent eventToSend)
        {
            var subscriberType = typeof (ISubscriber<>).MakeGenericType(typeof (TEvent));
            var subscribers = GetSubscribers(subscriberType);
            List<WeakReference> toRemove = new List<WeakReference>();

            foreach (WeakReference weakSubscriber in subscribers)
            {
                if(weakSubscriber.IsAlive)
                {
                    var subscriber = (ISubscriber<TEvent>) weakSubscriber.Target;
                    var syncContext = SynchronizationContext.Current;
                    if(syncContext == null)
                        syncContext = new SynchronizationContext();
                    syncContext.Post(s => subscriber.OnEvent(eventToSend), null);
                }
                else
                {
                    toRemove.Add(weakSubscriber);
                }
            }

            if(toRemove.Any())
            {
                lock (_syncObject)
                {
                    foreach (WeakReference weakReference in toRemove)
                    {
                        subscribers.Remove(weakReference);
                    }
                }

            }
        }
    }

 public interface ISubscriber<in T>
    {
        void OnEvent(T message);
    }

    public class StatsCounter : ISubscriber<ScoreMessage>, ISubscriber<AllPlayersFinished>
    {
        public static readonly object _sync = new object();
        private Dictionary<string, int> _players;

        public StatsCounter()
        {
            _players = new Dictionary<string, int>();
        }


        public void OnEvent(ScoreMessage message)
        {
            lock (_sync)
            {
                var name = message.Sender.Name;
                Console.WriteLine(String.Format("{0} scored !", name));
                if(_players.ContainsKey(name))
                {
                    _players[name]++;
                }
                else
                {
                    _players.Add(name,0);
                }
            }
        }

        public void OnEvent(AllPlayersFinished message)
        {
            lock (_sync)
            {
                Console.WriteLine("Results : ");
                foreach (KeyValuePair<string, int> keyValuePair in _players)
                {
                    Console.WriteLine(String.Format("{0}\t scored {1} time(s)"),
                        keyValuePair.Key, keyValuePair.Value);
                }
                Console.ReadKey();
            }
        }
    }

    public class GameManager : ISubscriber<FinishMessage>
    {
        private readonly IEventAggregator _ea;

        private int _counter;

        public GameManager(IEventAggregator eventAggregator)
        {
            _ea = eventAggregator;
        }

        public void OnEvent(FinishMessage message)
        {
            _counter++;
            lock (StatsCounter._sync)
            {
                Console.WriteLine(String.Format("{0} Finished ! <-----",message.Sender.Name));
                Console.ReadKey();
            }
            if(_counter == 3)
                _ea.Publish(new AllPlayersFinished());
        }
    }

public interface IPlayer
    {
        string Name { get; set; }
    }

    public class Player : IPlayer
    {
        private IEventAggregator _ea;

        public int Sum { get; set; }
        public string Name { get; set; }

        public Player(IEventAggregator eventAggregator, string name)
        {
            _ea = eventAggregator;
            Name = name;
        }

        public void Play()
        {
            Random r = new Random();
            while (Sum < 1000)
            {
                int num = (r.Next() % 20);
                Sum += num;
                if(Sum % 10 == 0)
                    _ea.Publish(new ScoreMessage{Sender = this});
                Thread.Sleep(10*num);
            }
            _ea.Publish(new FinishMessage() {Sender = this});
        }
    }

public class ScoreMessage
    {
        public IPlayer Sender { get; set; }
    }

    public class FinishMessage
    {
        public IPlayer Sender { get; set; }
    }

    public class AllPlayersFinished
    {
        
    }

class Program
    {
        static void Main(string[] args)
        {
            IEventAggregator eventAggregator = new SimpleEventAggregator();
            var player1 = new Player(eventAggregator, "Player 1 [POL]");
            var player2 = new Player(eventAggregator, "Player 2 [USA]");
            var player3 = new Player(eventAggregator, "Player 3 [RUS]");
            StatsCounter counter = new StatsCounter();
            eventAggregator.Subscribe(counter);
            GameManager manager = new GameManager(eventAggregator);
            eventAggregator.Subscribe(manager);

            Thread t1 = new Thread(player1.Play);
            Thread t2 = new Thread(player2.Play);
            Thread t3 = new Thread(player3.Play);
            
            Console.WriteLine("The game has started !");

            t1.Start();
            t2.Start();
            t3.Start();
        }
    }