Determining invocation order

Question

I have a simple sequence that looks like this:

IObservable<Message> MessageReceived

Several classes in different modules and threads of the system subscribe to this sequence, without schedulers, like this:

MessageReceived.Subscribe(onMessageReceived);

And here is the problem: Message is a pooled type. After I parse it, it needs to be released so the pool can use it again. But it can't be released while some subscriber is still using it. And there's no way (that I know of) to schedule something to happen after all other subscribers finished handling the message. I would have liked to use something like:

MessageReceived.SubscribeLast(p_message => Pool.Release(p_message));

Is there any way to do something like that? Maybe an extension method I could write? I don't want to rely on subscription order, because those happen on different threads and the timing is delicate. Also, there are timeouts and resubscriptions during application run.

Thanks in advance,

Lena

Answer 1

There is nothing in Rx that supports your requirement out of the box (afaik).

I imagine that you could wrap your object in a pool aware structure, or your object itself can encapsulate the pooling (like a SqlConnection object).

An alternate option may to have a Circular buffer of pooled objects that is large enough to assume that by the time you do a full loop that you can safely reuse the pooled object.

HTH, if not I would google object pooling or check Stackoverflow for patterns there.

Lee

---

Lee Campbell http://LeeCampbell.blogspot.com