Generic Payload for Prism Event Aggregator

Question

I am developing an WPF application and tried to design something event driven using Prism's Event Aggregator.

Currently I am trying to implement something like event queue for Prism events. To do that I want to subscribe related events and pass them to same method but Event Aggregator wants those methods to have same signature with the event.

Example events:

public class TestEvent1 : PubSubEvent<Class1>
{
}
public class TestEvent2 : PubSubEvent<Class2>
{
}
public class TestEvent3 : PubSubEvent<List<Class3>>
{
}
public class TestEvent3 : PubSubEvent<string>
{
}

Subscriptions:

        _eventAggregator.GetEvent<TestEvent1>().Subscribe(OnTestEvent1, true);
        _eventAggregator.GetEvent<TestEvent2>().Subscribe(OnTestEvent2, true);
        _eventAggregator.GetEvent<TestEvent3>().Subscribe(OnTestEvent3, true);

Example callback method:

    private void OnTestEvent1(Class1 object1)
    {
        // do something      
    }

Since I only receive data when event published, I tried something like this to use as event payload type, but it doesn't look right:

// Payload
public interface IMessage
{
    public object Data { get; set; }
    public Type Datatype { get; set; }
    public PubSubEvent EventType { get; set; }
}

// Events
public class TestEvent1 : PubSubEvent<IMessage>
{
}
public class TestEvent2 : PubSubEvent<IMessage>
{
}
public class TestEvent3 : PubSubEvent<IMessage>
{
}

// Subscriptions
_eventAggregator.GetEvent<TestEvent1>().Subscribe(EventHandler, true);
_eventAggregator.GetEvent<TestEvent2>().Subscribe(EventHandler, true); 
_eventAggregator.GetEvent<TestEvent3>().Subscribe(EventHandler, true);

// Callback
private void EventHandler(IMessage payload)
{
    // do something      
}

Is this viable and how can I improve or change my "generic" payload?

Extra information: In my scenario I have multiple UDP servers that periodically receives new data, deserialize it to objects and publish events. Related "managers" subscribed to those events and get triggered when new data received.

My goal is try to implement an event queue like system in my event receiving classes so it will be easier to deal with multithreading issues.

Here are some diagrams that may help me explain myself better:

My architecture:

My "event queue"

Answer 1

When you publish an event with a reference to an object, consider these potential problems:

• Any listener will be dependent on the type/assembly of the event parameter.
• The referenced object will be held for some time. For how long? You don't know.
• The referenced object could have been altered/disposed.
• Listeners might handle the referenced object on a different thread.

After dealing with PubSubEvent<T> for several years now, I believe there is only one suitable pattern that successfully handles all cases. Publish your event with a unique identifier, e.g. a Guid.

public class MyItemAddedEvent : PubSubEvent<Guid> {}

Then inject a provider wherever you listen for this event:

public class SomeListener
{
    private readonly IMyItemProvider myItemProvider;

    [ImportingConstructor]
    public SomeListener(IEventAggregator eventAggregator,
                        IMyItemProvider myItemProvider)
    {
        this.myItemProvider = myItemProvider;

        eventAggregator.GetEvent<MyItemAddedEvent>().Subscribe(OnMyItemAdded, true);
    }

    private void OnMyItemAdded(Guid id)
    {
        var myItem = myItemProvider.Get(id);
        // Do stuff
    }
}

Now it is the responsibility of the provider class to deliver a valid and up-to-date object given a unique id.

Answer 2

This is more of a code-review question, isn't it?

I would very much prefer strong typing, though, and I wouldn't want to build a queue around the event aggregator, that will always be tedious because you never know when new event types show up. Instead, I'd build my own event aggregator with queueing built in (starting from the existing one).

Also, I'd look into dataflow, for example, because the basic working mode of the event aggregator (fire and forget, one sender, multiple or no receivers) doesn't seem to work well with queueing. If you queue at the sender-side, to you wait for one receiver or all? Do you queue when there are no receivers or do you discard then? If you queue at the receiver-side, why bother at all? The receiver can implement the queue on its own.