How to implement asynchronous observer pattern or events

Question

I'm looking for a way to implement the following:

A says: "Yo, X happened. Bye."

Others see that and start doing some work.

In other words, I would like to fire an event and let others handle that in a fire and forget way.

So I've looked into the observer pattern: https://msdn.microsoft.com/en-us/library/dd783449(v=vs.110).aspx. However this example is synchronous, and if the observers take a long time to do their work, the notify method blocks for a long time.

I also looked at how to raise events: https://msdn.microsoft.com/en-us/library/9aackb16(v=vs.110).aspx. Also this example is synchronous, and blocks the sender for a long time when the handler takes long to handle the event.

My question is:

How do I do fire and forget events/messages/delegates in C#?

Possible duplicate of [How do I make an eventhandler run asynchronously?](http://stackoverflow.com/questions/1916095/how-do-i-make-an-eventhandler-run-asynchronously) — Tatranskymedved, Jan 25 '17 at 08:44
@Tatranskymedved But this dup is old and it's an absolutely outdated solution... — Matías Fidemraizer, Jan 25 '17 at 08:51

Matías Fidemraizer · Accepted Answer · 2017-01-25T10:21:27.640

3

Probably you should meet Task Parallel Library (TPL) Dataflows. There's one data flow called ActionBlock<TInput> that should be a good start for you:

The ActionBlock<TInput> class is a target block that calls a delegate when it receives data. Think of a ActionBlock<TInput> object as a delegate that runs asynchronously when data becomes available. The delegate that you provide to an ActionBlock<TInput> object can be of type Action or type System.Func<TInput, Task>[...]

Therefore, what about giving a Func<TInput, Task> to ActionBlock<TInput> to perform asynchronous stuff? I've modified the sample found on this TPL Dataflow MSDN article:

List<Func<int, Task>> observers = new List<Func<int, Task>>
{
      n => Console.WriteLine(n),
      n => Console.WriteLine(n * i),
      n => Console.WriteLine(n * n / i)
};

// Create an ActionBlock<int> object that prints values
// to the console.
var actionBlock = new ActionBlock<int>
(
    n => 
    {
         // Fire and forget call to all observers
         foreach(Func<int, Task> observer in observers)
         {
              // Don't await for its completion
              observer(n);
         }         
    }
);

// Post several messages to the block.
for (int i = 0; i < 3; i++)
{
   actionBlock.Post(i * 10);
}

// Set the block to the completed state
actionBlock.Complete();

// See how I commented out the following sentence.
// You don't wait actions to complete as you want the fire
// and forget behavior!
// actionBlock.Completion.Wait();

You might also want to take a look at BufferBlock<T>.

edited Jan 25 '17 at 10:21

answered Jan 25 '17 at 09:05

Matías Fidemraizer

63,804
18
124
206

Hmm, is it invokes delegate in different threads? I got same for ```n =>Thread.CurrentThread.ManagedThreadId``` delegate. – tym32167 Jan 25 '17 at 09:33
@tym32167 It's part of *Task Parallel Library*. It really parallelizes work in multiple CPU cores. – Matías Fidemraizer Jan 25 '17 at 09:40
@tym32167 See this article https://msdn.microsoft.com/en-us/library/mt604413(v=vs.111).aspx – Matías Fidemraizer Jan 25 '17 at 09:41
@tym32167 I wouldn't be very concerned about spawning threads. See how proven techs like NodeJS or Redis are single-threaded and can delivery work at the light speed. With Dataflows you've even a more powerful tool, because it can spawn physical threads and, depending on the scenario, it might outperform logical threads. – Matías Fidemraizer Jan 25 '17 at 09:46
Aha, I was not aware about max degree of parallelism. Thanks, very informative. Good to know. – tym32167 Jan 25 '17 at 09:48
Yeach, I know that asynchronous and parallel is different things. I just thought about parallel processing events and work with exceptions. So its looks like clear for me now, thank you :) – tym32167 Jan 25 '17 at 09:52
@tym32167 No problem! – Matías Fidemraizer Jan 25 '17 at 10:00
This doesn't really look like I can add more observers to it. Or am I not understanding it correctly? – user369117 Jan 25 '17 at 10:09
@user369117 wouldn't be as easy as maintaining a delegate list and call one by one in the delegate you pass to `ActionBlock` constructor? – Matías Fidemraizer Jan 25 '17 at 10:13
@user369117 BTW this is a very simple solution. Here's an article on which someone explains how to implement a full fledged producer-consumer pattern using `BufferBlock` (another TPL data flow): http://blog.stephencleary.com/2012/11/async-producerconsumer-queue-using.html – Matías Fidemraizer Jan 25 '17 at 10:15
@user369117 See my edited answer where I've added an observer list. – Matías Fidemraizer Jan 25 '17 at 10:21
nice one. And thanks for the link to Stephen Cleary's blog. – user369117 Jan 25 '17 at 10:53
@user369117 I believe that now you got the tool, and it's up to you the *how to improve it*. Maybe this goes beyond of what you want, but do you know [ZeroMQ](http://zeromq.org/)? – Matías Fidemraizer Jan 25 '17 at 10:57

How to implement asynchronous observer pattern or events

1 Answers1