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.

Answer 1

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.

Answer 2

It all depends on how you think of your observables.

If it's a stream of events independent of the observers, you can just expose the observable to anyone that want's to subscribe to.

That was the approach I followed on my reactive GeoCoordinateWatcher. The wrapped GeoCoordinateWatcher class will generate events independent of the subscription.

For the reactive Geolocator I chose to follow the same approach. But because the Geolocator needs parameterization to produce events, I could have chosen to implement an observable factory.

The bottom line is (as @Christopher said) that you send commands to something that exposes an observable and not to the observale itself.

You migth do something like sending commands to the observable by applying Rx operators. If those filters are to be applied remotely, you can implement (as @Christopher said) an IQbservale instead.