[louthy]

That implies a two way dispatch which is not in the remit of reactive surely? A system that's closer to Haskell's pipes is probably what you'd want.

I think ISubject<I,O> has the interface for a pipe, with IObservable being a Producer equivalent, and IObserver being a Consumer equivalent. So there may be some cunning that could be done with your own implementations [to create an Effect]. (This is just wild brainstorming without looking at the code, so take all of this with a large pinch of salt).

Working within the Rx system, there'd be two ways of doing this:

The Observable has a mechanism for slowing down, but obviously it can't get instruction from observers - so it would have to make a judgment on what is 'too much'; a slow down would affect all observers. Definitely (well probably) not what you want.

The other way is to use the various buffering functions in Rx on the subscription, or roll your own function that has some intelligence. That localises the 'backing up' on a per observer basis, but doesn't slow down the observable itself.

To 'Stop' you could switch your subscription to Observable.Never<T> until you're ready to receive messages. Obviously you'll miss messages generated on the other stream whilst you were not listening.

[platz]

Reactive = resilient + eventbased + responsive + scalable

I think it is relevant, what allows you to ignore the behavior of the system in favor of a component?

Dean wampler in a recent talk spoke of "Reactive Streams" which add this capability, but I have not seen it yet.

I have not seen anyone do this in Rx yet. Also ideally the back pressure is out of band to avoid starvation of that channel

[louthy]

I guess have an 'out of band' observable? Each observer that needs to communicate instruction back to the original observable can drop messages into an OoB observable, the original observable can subscribe to the OoB observable and then react to those messages as it sees fit.

Or do you think it should be an inherent feature of the stream itself? Isn't the point of IObservable to be the dual of IEnumerable, so calling Select on IEnumerable is a projection from A -> B, calling Select on IObservable is also a projection from A -> B, I think having an additional effect breaks that duality. So an Observable that has a side-channel or out-of-band stream wouldn't be an Observable, it'd be another category (which is why I suggested Pipes).

I totally understand why you'd want something like that, I'm just not sure it's fair to suggest Rx has design mistakes because of this though.

[platz]

Well, speaking to the point of duality, I think Rx solves the problems it sets out to solve very nicely, and I would not say it has "design mistakes".

It certainly seems possible to use Rx primitives to build more functionality, yes.

Though everyone will have to build this themselves (if, of course, they even need it), and they'll probably all build it differently, where other lib's have this concept baked in (i.e. Pipes).

IMHO, libraries shouldn't try to do everything anyway, and to that end, Rx does it's thing and does it well (unix philosophy), as it's based on these mathematical abstractions.