11
\$\begingroup\$

I'd like to get input on my approach of continuously receiving messages from an Azure Service Bus queue using the async part of the library. My main concern being whether its "safe" to use Task.Factory.StartNew to process the incoming messages and then continue to call the Action's to continue receiving new messages.

The processor class is started like this:

public void Start()
{
    ThrowIfDisposed();
    lock (_lockObject)
    {
        Trace.WriteLine(string.Format("Started '{0}' allocation.", _projectSettings.Name));

        _cancellationSource = new CancellationTokenSource();
        Task.Factory.StartNew(ReceiveAllocationMessages, _cancellationSource.Token);
    }
}

And then the actual continuous receival of messages from the (Azure Service Bus) QueueClient (from the Microsoft.ServiceBus.Messaging namespace):

private void ReceiveMessage()
{
    Task.Run(async () =>
    {
        //Receive a new message async within 1 minute
        return await _queueClient.ReceiveAsync(TimeSpan.FromMinutes(1));
    })
    .ContinueWith(t =>
    {
        if (t.Result != null)
        {
            //The result is not null so we process the Brokered Message
            Task.Factory.StartNew(() =>
            {
                ProcessMessage(t.Result)
                ReceiveNextMessage();
            }, _cancellationSource.Token);
        }
        else
        {
            // Continue receiving and processing 
            // new messages until told to stop.
            ReceiveNextMessage();
        }
    });
}

private void ReceiveNextMessage()
{
    if (_cancellationSource.IsCancellationRequested == false)
    {
        //Continue to receive new messages every 1 minute
        Task.Delay(TimeSpan.FromMinutes(1))
            .ContinueWith(t => ReceiveMessage());
    }
}

When the process is stopped or disposed the Stop method is called in order to cancel the CancellationTokenSource as shown here:

public void Stop()
{
    lock (_lockObject)
    {
        using (_cancellationSource)
        {
            if (_cancellationSource != null)
            {
                _cancellationSource.Cancel();
                _cancellationSource = null;
            }
        }

        Trace.WriteLine(string.Format("Stopped '{0}' allocation.", _projectSettings.Name));
    }
}

Does this seem like a reasonable way to go about implementing continuous receival of messages in an async fashion?

UPDATE: I posted a gist with the code from above polished and put into a class. Hopefully improves readability Gist of the queue Processor class

\$\endgroup\$
  • 1
    \$\begingroup\$ Indent your code properly, please. ._. \$\endgroup\$ – kleinfreund Feb 17 '14 at 9:37
  • \$\begingroup\$ How is this continuous? As far as I can see, if you receive one message, you process it and stop, you don't try to receive another message. \$\endgroup\$ – svick Feb 17 '14 at 9:37
  • \$\begingroup\$ @kleinfreund I updated the code, so hopefully its a bit better now. \$\endgroup\$ – sitereactor Feb 18 '14 at 9:35
  • \$\begingroup\$ @svick Thanks for pointing that out. I had left out a ReceiveNextMessage call within the ContinueWith. Should be updated in the code sample. \$\endgroup\$ – sitereactor Feb 18 '14 at 9:37
4
\$\begingroup\$

First, some notes:

  1. You shouldn't be using Task.Factory.StartNew() unless you have to, Task.Run() is better.
  2. I think you're using Task.Run() (and Task.Factory.StartNew()) and ContinueWith() too much. Only use Task.Run() when you actually need to run the code on another thread. And ContinueWith() is almost always better written using await.

    If you're using it to support cancellation, you can just use ThrowIfCancellationRequested() instead.

  3. Don't use == false, ! is much clearer.
  4. Your your code, the processing can be Started two times. You should probably guard against that.
  5. Your Start() and Stop() methods seems to be written with support for restarting in mind. It might be simpler to forbid that.

Now, about the structure of your code: I think a much clearer way to write your code would be to use a loop, instead of recursion. That way, your top level method can be made very clear:

while (!_cancellationSource.IsCancellationRequested)
{
    await ReceiveMessageAsync();
    await Task.Delay(TimeSpan.FromMinutes(1));
}

The ReceiveMessageAsync() method could then look like this, which is a lot shorter and much more readable:

private async Task ReceiveMessageAsync()
{
    var message = await _queueClient.ReceiveAsync(TimeSpan.FromMinutes(1));

    if (message != null)
    {
        _cancellationSource.ThrowIfCancellationRequested();
        ProcessMessage(message);
    }
}

Note that this version behaves differently than yours with regards to exceptions. In your code, if an exception happens, it's either silently ignored (if it's in ReceiveAsync()) or it just silently stops receiving (if it's in ProcessMessage()). Both are very bad and make errors hard to debug. With my code, any exception stops receiving and is reported to the top-level Task.

You should then devise some way of reporting any exceptions on that Task.

A simple solution would be to store the Task in Start() and then rethrow any exceptions in Stop(). A cleaner solution might be to change void Start() into something like Task RunAsync() and report exceptions using the returned Task.

You could also consider not stopping if some specific exception happens (e.g the server is temporarily unavailable). But that should happen only for some specific exceptions and those exceptions should be logged anyway, to help with debugging.

\$\endgroup\$
  • \$\begingroup\$ Thanks for taking the time to write this review. Appreciate the addition of the link to Task.Factory.StartNew vs. Task.Run. I have started refactoring the code based on some feedback, which I got through the gist on github, which is very much in-line with what you are suggesting. I'll post back with a link to the gist when I've done a bit more testing with this alternative approach. \$\endgroup\$ – sitereactor Feb 19 '14 at 11:44
  • 1
    \$\begingroup\$ #3 is debatable. " if (!IsAvailable)" vs " if (IsAvailable == false" - its hard to see at times and has led to more than one bug. \$\endgroup\$ – StingyJack Oct 12 '16 at 17:02
  • \$\begingroup\$ Agreed @StingyJack I hope none of my teammates see number #3! I think == false is a much more readable. I'd throw in that you should name your variables in a way that there isn't too much == false or ! involved, try to stay in the affirmative sense \$\endgroup\$ – Jacob McKay Nov 14 '17 at 18:52

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.