Restart an observable subscription when a property changes

user_136525 0
Viewed 590

I have a simple subscription that is created in MyClass:

//myService.Connect returns IObservable<MyData>
myService.Connect(requestParameters)
         .Subscribe(DoSomething);

There is a property in MyClass that I am listening to which is used to create the requestParameters. My goal is to call myService.Connect again whenever that property changes, using the new requestParameters, and unsubscribing from the previous subscription.

Is there a straightforward way to do this with RX? I had looked at Switch but that is used when subscribed to an Observable that emits Observables. I also had a look at TakeUntil, and I can keep the subscription alive until the property changes, but I am not sure how I can trigger an automatic resubscribe.

At the end of the day I can just call myService.Connect myself whenever the property changes but wanted to see if there was some existing RX functionality that I can use instead.

system.reactive c#
1 Answers

Depending on what exactly you want to do, the Switch operator might indeed what you are looking for. It can be used to provide one simple observable to the public where the underlying source observable can be "switched" to a different observable. As long as the type is the same it can work. In fact, the Switch() operator requires that the observables must be of the same type. See the following image from the documentation above:

Marble diagram of the Switch operator Marble diagram of the Switch operator from http://reactivex.io

The top part shows observables which are active and sending values to the "visible" observable at the bottom. But at some point the original source is changed or "switched" (circles to triangles), while the "visible" observable at the bottom produces values like nothing happened.

In your case you get a new observable with your myService.Connect(requestParameters). You fill this value to a subject of IObservable<YourType> via the OnNext() method. Check the following example:

ISubject<IObservable<int>> sourceOfObservables = new Subject<IObservable<int>>();

IObservable<int> publicSource = sourceOfObservables.Switch();

IDisposable testSubscription = publicSource.Subscribe(it => {
    Console.WriteLine("Value received: "+it);
});

sourceOfObservables.OnNext(Observable.Range(1, 5));
sourceOfObservables.OnNext(Observable.Range(10, 5));
sourceOfObservables.OnNext(Observable.Range(100, 5));

This will generate the following output:

Value received: 10
Value received: 11
Value received: 12
Value received: 13
Value received: 14
Value received: 100
Value received: 101
Value received: 102
Value received: 103
Value received: 104

As you see, you have only one active subscription testSubscription where you can read the 15 different values. However they come from three separated sources/observables. This is similar to Concat where you can concatenate observables, but in this time you are kinda "hot swapping" one active subscription with another, without everyone need to resubscribe to your new obserable.

You can solve you problem by first defining an ISubject<IObservable<YourType>>. Then when your requestParameters changes you push a new IObservable<YourType> instance from your myService.Connect() method to your ISubject<IObservable<YourType>> instance with the OnNext() method (as seen in the example). From now on the next values come from that observable.

Bonus:

When requestParameters is already from an observable, you can use the Select() operator (from System.Reactive.Linq, not the normal LINQ) to build the IObservable<YourType> automatically, which you can "flatten out", "pseudo concat" or literal "switch" with Switch():

IObservable<string> requestParametersChangeStream;

requestParametersChangeStream             // produces "string" objects 
    .Select(it => myService.Connect(it))  // produces "IObservable<YourType>" objects
    .Switch()                             // make it so it looks like
                                          // an IObservable<YourType>
    .Subscribe(it => DoWhateverYouWant(it));
user_2869340
Related