Two way communication with Reactive Extensions

Question

Let's imagine I have a hot observable that is a source of weather events. This source is a socket connection to a remote server that provides information about the weather in my current location.

That remote server can also send me events about other related topics as traffic, extreme weather warnings, etc... if I send a command to indicate this desire.

How can I model this with Reactive Extensions without creating coupling between observable and observer?

The idea is that when I subscribe ExtremeWeatherWarnings (observer) to my WeatherServerConnection (observable), somehow the first issue some commands to the second, so it enables the subscription.

I have tried to implement an special interface in the observer, that tells the observable the commands it needs to execute in subscription and unsubscription, but it does not work in the moment I put a Where in the middle because LINQ RX is wrapping the observable, and that wrapper does not implement any interface.

I can also require an instance of the WeatherServerConnection object on the ExtremeWeatherWarnings constructor, but that will create coupling and I would like to avoid that.

Cheers.

Looks like you need something similar to ISubject. Check out Ping-Pong example — Dmitry Sevastianov
That example is very naive. I need that ExtremeWeatherWarnings populate other kind of things for its consumers. Thanks! — vtortola

cwharris cwharris · Accepted Answer · 2014-03-31T20:47:37

If your observables is designed to send generic messages, and your observer is design to translate them, you also need a way of indicating to the producer what kinds of messages you're interested in.

One way to do this is "requesting an observable".

ObservableWeatherSource GetNotifications(WeatherWarnings warningTypes, string location);

Another way might be to lazily indicate what notifications you're interested in.

ObservableWeatherSource source = GetWeatherSource();

source
    .Where(x => x.WeatherWarningType === WeatherWarnings.Rain)
    .Subscribe(new WeatherObserver());

source.ExpressInterestIn(WeatherWarnings.Rain, "San Francisco");

Or, possibly, you might be interested in writing a specialized query language for weather. You could probably do this through the IQbservable and a query provider, but I have little knowledge of this area of Rx.

Two way communication with Reactive Extensions

2 Answers