4
\$\begingroup\$

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

\$\endgroup\$
  • \$\begingroup\$ Could you please clarify what the requirements on the order of execution of events are? Your code currently executes short running tasks synchronously by the order they were added to the 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 the BlockingCollection. Is this a requirement, or is this an implementation detail? \$\endgroup\$ – Martin Jan 2 '16 at 22:55
  • \$\begingroup\$ @Martin please see my update \$\endgroup\$ – Yoav Jan 3 '16 at 9:12
  • \$\begingroup\$ So, when _ioService.IoEvent raises the event twice with two different int portNum the event handlers can be called simultaneously but when it raises the event twice with the same int portNum the event handlers should be executed one by one? \$\endgroup\$ – Martin Jan 3 '16 at 10:28
  • \$\begingroup\$ @Martin correct \$\endgroup\$ – Yoav Jan 3 '16 at 10:29
  • \$\begingroup\$ Is there a different method for each 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\$ – Martin Jan 3 '16 at 12:22
3
+50
\$\begingroup\$
  1. The method StartListening takes one argument object dummy. It is not used in the method and from its name I conclude that it is not intended to be used. It is possible to declare the method with no arguments and call a variant of Task.Factory.StartNew that takes a parameter of type Action and TaskCreationOptions to start the new task.

  2. The field private static EventWaitHandle _eventWaiter is only initialized in the constructor and never used. Its purpose is unclear from the code.

| improve this answer | |
\$\endgroup\$
1
\$\begingroup\$

IoService event

I would suggest having an event handler for that iOservice event, and in that event handler you would have a dicitionary collection that will hold for example : "Port80" : WebHandler "Port21" : FtpHandler

this way you would initialize this collection or load it from a configuration file with all the required port numbers to handle. Of course you can handle them inside those classes with a task that receives your cancelation token from the main class calling them.

| improve this answer | |
\$\endgroup\$

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Not the answer you're looking for? Browse other questions tagged or ask your own question.