System.Reactive, processing azure queue message one at a time

Question

Having a little trouble with this one and hoping you guys can help... I am using C# System.Reactive. I have one observable that is simply an interval that gets azure queue messages from an azure storage queue, one at at time. This looks like this:

public IObservable<QueueMessage> On(
    string queueName, 
    int poleIntervalSeconds = 5
)
{
    return Observable
        .Interval(TimeSpan.FromSeconds(poleIntervalSeconds))
        .SelectMany(async _ =>
        {
            IAzureQueueAdapter queueAdapter = ConnectToQueue(queueName);
            Response<QueueMessage[]> response = await queueAdapter.ReceiveMessagesAsync(1);
            return response.Value;
        })
        .SelectMany(messages => messages)
        .Do(async queueMessage =>
        {
            IAzureQueueAdapter queueAdapter = ConnectToQueue(queueName);
            await queueAdapter.DeleteMessageAsync(queueMessage.MessageId, queueMessage.PopReceipt);
        });
}

// Usage

IObservable<ProcessingCompletedEvent> processingCompletedEvent = eventAggregator
    .On<ProcessingCompletedEvent>(); // Not the same On

IObservable<QueueMessage> queueMessageObservable = queueAdapter.On(queueName, 5);

It's fairly straight forward, and for straight async processing of queue messages (windows service app) this works perfectly. However I now have an external process that uses a third party C++ SDK that I have to use pinvoke to execute. The SDK basically exposes event delegates that I have wrapped with observables using FromEventPattern. This all works fine for normal async stuff as I said, but the problem is I now have a subscription that needs to listen to the queue and ALSO wait for a completion event from the SDK wrappers observables.

In simple terms, I need to only get a new message off the queue if the previous message also has a matching completed event. This is due to issues with the SDK, which crashes if I try to do two things at a time.

For reference here is the completed event that gets emitted:

public class ProcessingCompletedEvent : EventMessage<QueueMessage> 
{
     public ProcessingCompletedEvent(QueueMessage data) : base(data) { 
        MessageId = data.MessageId; 
     }
     public string MessageId { get; set; }
     public string? CorrelationId { get; set; }
}

To be clear I don't really have an option to change the source of the messages to service bus or anything. However, I am fine with messages backing up locally if another one is retrieved while the previous one is still being processed, I just want it to wait to process it until the previous message finished processing. I do have the option to change the On or create a new function to it with or without the interval if it better serves the purpose, but I do feel like the interval is necessary.

Here's a hint to help. Never use SelectMany(async _ => ...). Use SelectMany(_ => Observable.FromAsync(() => ...)). — Enigmativity
– Enigmativity, Commented Jun 1, 2024 at 12:15

Enigmativity · Accepted Answer · 2024-06-01 12:21:05Z

I've refactored out the async lambdas (as they don't play well with Rx) to get you this:

public IObservable<QueueMessage> On(string queueName, int poleIntervalSeconds = 5) =>
    Observable
        .Defer(() =>
        {
            IAzureQueueAdapter queueAdapter = ConnectToQueue(queueName);
            return
                from n in Observable.Interval(TimeSpan.FromSeconds(poleIntervalSeconds))
                from rqms in Observable.FromAsync(() => queueAdapter.ReceiveMessagesAsync(1))
                from qm in rqms.Value
                from u in Observable.FromAsync(() => queueAdapter.DeleteMessageAsync(qm.MessageId, qm.PopReceipt))
                select qm;
        });

I can't be sure it works, but hopefully it's a little closer.

If you need to control the thread that's being used more closely then try using EventLoopScheduler. It's a scheduler that spins up its own thread, so can't have two things running at once.

Collectives™ on Stack Overflow

System.Reactive, processing azure queue message one at a time

1 Answer 1

Comments

Your Answer

Hot Network Questions

Collectives™ on Stack Overflow

1 Answer 1

Comments

Your Answer

Sign up or log in

Post as a guest

Related