I have a service that raises multiple events, some of them can be raised at the same time. I need to process those events and run a potentially long running method based on the event arguments.
What I did is to create a BlockingCollection<T>
that will store the events and a Task that will keep taking one event at a time until it will be signaled to stop using a CancellationTokenSource
.
This is my code:
public class EventsTest
{
//private fields
private BlockingCollection<int> _queue; //a concurrent collection to hold incoming events
private CancellationTokenSource _tokenSource;
private IoService _ioService; //a service that raises multiple events
private static EventWaitHandle _eventWaiter; // used for waiting inside a method
public EventsTest()
{
_queue = new BlockingCollection<int>();
_tokenSource = new CancellationTokenSource();
_eventWaiter = new EventWaitHandle(false, EventResetMode.AutoReset);
_ioService = new IoService();
_ioService.IoEvent += _ioService_IoEvent;
//Start Listening to IO events
var t = Task.Factory.StartNew(StartListening, _tokenSource, TaskCreationOptions.LongRunning);
}
//IO events listener
private void _ioService_IoEvent(byte[] desc, int portNum)
{
//add events to a blocking collection
_queue.Add(portNum);
}
//a while loop running on a separate task
//and taking the event one by one
private void StartListening(object dummy)
{
while (!_tokenSource.IsCancellationRequested)
{
try
{
var eve = _queue.Take(_tokenSource.Token);
switch (eve)
{
case 0:
//run the required method on a separate task so it won't block the loop
Task.Factory.StartNew(LongRunningMethod);
break;
case 1:
//invoke a potential blocking method
break;
default:
break;
}
}
catch (OperationCanceledException)
{
// _tokenSource was signaled to stop. do nothing
}
}
}
static readonly object _locker = new object();
//sample long running method
private void LongRunningMethod()
{
lock (_locker)
{
Thread.Sleep(5000);
}
}
}
Am I on the right track here? what can I do in order to make it more robust?
UPDATE
The process of this class is as follows:
- Listen to events raised by a remote service
- Handle each event based on it's event arguments in a predefined method
- A method execution time can be long
- Different methods can run simultaneously (That's why I start each method with the
Task.Factory.StartNew(LongRunningMethod);
) - If an event handling method needs to run again when the previous one wasn't finished it should wait for it (that's why I placed the
lock
statement in the method implementation)
please note that I originally posted this question here but I guess that it's more suitable for this site
BlockingCollection
. Long running tasks are being executed concurrently with short running tasks, but one at a time and not necessarily in the order they were added to theBlockingCollection
. Is this a requirement, or is this an implementation detail? \$\endgroup\$_ioService.IoEvent
raises the event twice with two differentint portNum
the event handlers can be called simultaneously but when it raises the event twice with the sameint portNum
the event handlers should be executed one by one? \$\endgroup\$int portNum
that needs to be called (one method per a value) or is there only one method that needs to be called with a parameter? \$\endgroup\$