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