Event-based producer/consumer in C# I want to create a production-ready producer/consumer that could help me avoid thread synchronization hell. Is this thread-safe? The main issue is to be safe with exceptions that can arrive.

public class AsynchSimpleProducer&lt;T&gt; : IDisposable
{
    private readonly Action&lt;Exception&gt; error;
    private readonly BlockingCollection&lt;T&gt; blockingCollection = new BlockingCollection&lt;T&gt;(50);

    public event Func&lt;T, Task&gt; NewItem;

    public AsynchSimpleProducer(Action&lt;Exception&gt; error)
    {
        this.error = error;

        Task.Factory.StartNew(() =&gt;
                              {
                                  //this loop is ends only when blockingCollection.CompleteAdding called
                                  Parallel.ForEach(blockingCollection.GetConsumingPartitioner(), SendItem);
                              });
    }

    private void SendItem(T item)
    {
        try
        {
            if (NewItem !=null)
                NewItem(item).Wait();
        }
        catch (Exception ex)
        {
            error(ex);
        }
    }

    public void Send(T newValue)
    {
        blockingCollection.Add(newValue);
    }

    public void Dispose()
    {
        blockingCollection.CompleteAdding();
    }
}


Example of usage:

simpleProducer = new AsynchSimpleProducer&lt;int&gt;(Error);
simpleProducer.NewItem += simpleProducer_NewItem;
//this part can be in task
var next = random.Next(0, 1000);
simpleProducer.Send(next);
I don't really understand how you want to use this so I'm just going to add a couple of comments for you to ponder.

Naming

It's more common to abbreviate Asynchronous to Async not Asynch when used in identifers: AsynchSimpleProducer&lt;T&gt; -> AsyncSimpleProducer&lt;T&gt;.

NewItem isn't a great name for an event, events generally need a verb to make sense e.g. Closed, Clicked, Closing are all good names for an event. Consider something like ItemAdded or similar.

Thread safety

You aren't raising your event safely:

if (NewItem != null)
    NewItem(item).Wait();


NewItem can change between the check and using it. If a listener unsubscribes in this time you'll get an Exception! The proper way to do this is:

var newItem = NewItem;
if (newItem != null)
{
    newItem(item).Wait();
}


Note that I have also added in braces as I personally think they should always be included, even for single line blocks.

Asynchronicity

I had to type something that made me very sad.

newItem(item).Wait();


It's not async if you immediately wait on it. It would be interesting to know what sort of work you're passing in here as it's not obvious whether you need to try to make this asynchronous or not. I'm concerned that you have confused threading and asynchronicity.

Edit

I can't believe I didn't mention this before, I must have been distracted...

If your event handlers throw (the things listening on the NewItem event), you're toast. Say you have 2 listeners, if the first one throws, the second wont be called.

You need to go through each one in the invocation list:

 // untested...
 var runningTasks = new List&lt;Task&gt;();
 foreach (var delegate in newItem.GetInvocationList())
 {
     runningTasks.Add(delegate(item));
 }
 // catch exceptions here.
 Task.WaitAll(runningTasks.ToArray());


I wish I could suggest something useful but I still haven't figure out what you intend on doing with this thing.