response-gating next message send, with Rx

Question

given a List<Message> i send out the first message with my Send(message). Now I would like to wait for (an asynchronous) response to come back before i send out the next message...

Block until notified 'old' way

i know how to implement an event-based solution for this situation, using thread locking / with Monitor.Wait and Monitor.Pulse

Reactive 'new' way?

But I was wondering whether it would make sense to utilize Reactive Extensions here?

If Rx would convey worthwhile benefits here then how could I make the response reactively gate the next send invocation? Obviously it would involve IObservable, probably two as primary sources, but then what?

Answer 1

The question is not very specific and seems to be very general in the sense that you have not mentioned what is the sender receiver etc, so the answer would also be very general :)

var receiveObs = //You have created a observable around the receive mechanism 
var responses = messages.Select(m => {
   send(m);
   return receiveObs.First();
}).ToList();

Answer 2

I think Rx is a good choice here, but I think I could be missing something in your requirements. From what I understand Rx provides a very simple solution.

If you already have a list of messages then you can send them reactively like so:

messages
    .ToObservable()
    .ObserveOn(Scheduler.ThreadPool)
    .Subscribe(m =>
    {
        Send(m);
    });

This pushes the calls to Send to the thread-pool and, by the built-in behaviour of observables, each call to Send waits until the previous call is completed.

Since this is all happening on a different thread your code is non-blocking.

The extra benefit of Rx is that you wouldn't need to change the behaviour or signature of your Send method to make this work.

Simple, huh?

I tested this and it worked fine given my understanding of your problem. Is this all you need or is there something I missed?