How to determine that the switch has changed its output?

Question

I have a data source with the following header:

IObservable<IObservable<object>> MagicSource(IObservable<string> strings) 

I click the lines there and give me the results, but taking into account that I only need answers for the last line, I use the Switch statement to undo the previous ones:

 MagicSource(StringsProvider).Switch().Subscribe(r => {/*...*/}); 

This works well, but I also need to be notified that the string currently being processed has changed in order to prepare my presentation of the results. How to get such a notification?

Answer 1

I personally would avoid Do statements.

You could create a projection.

In this projection, I assume that you want to take your action to delete the current internal sequence (perhaps clear current records from the table?)

 MagicSource(StringsProvider) .Select(inner=>inner.Finally(CleanUp)) .Switch() .Subscribe(r => {/*...*/}); 

If you want to do the same, but aim at subscribing (rather than deleting) the internal sequence, then you will need a little more work. I would create a symmetric operator, finally Initially(Action)

 public static class ObservableEx { public static IObservable<T> Initially<T>(this IObservable<T> source, Action onSubscribe) { return Observable.Create<T>(o=>{ try { onSubscribe(); return source.Subscribe(o); } catch (Exception ex) { o.OnError(ex); return Disposable.Empty; } }); } } 

And then you can use it like this:

 MagicSource(StringsProvider) .Select(inner=>inner.Initially(Prepare)) .Switch() .Subscribe(r => {/*...*/}); 

Answer 2

The easiest way is to simply use the Do statement:

Here are a few varieties.

If you don't know anything about how MagicSource works and cannot accurately associate new incoming lines with new internal streams:

 Action NewStreamProduced = ...; MagicSource(StringsProvider).Do(_ => NewStreamProduced).Switch().... 

If you know that MagicSource will synchronously create a new internal thread every time a new line arrives, you can put your Do before calling MagicSource :

 Action<string> NewStringArrivedWhichMeansWeAreSwitching = ...; MagicSource(StringsProvider.Do(NewStringArrivedWhichMeansWeAreSwitching)).Switch()... 

Answer 3

You must add an index to each element indicating where it came from. Pretty simple to do.

 IObservable<Tuple<string, int>> results = MagicSource(StringsProvider) .Select((v,i)=>v.Select(y=>Tuple.Create(y,i)) .Switch() 

Answer 4

In fact, the best way is as follows. You need to use multicast.

 // Ensure that you are using multicast var source = MagicSource(strings).Publish().RefCount(); // Subscribe to this for notification and count var notifications = source.Select((v,i)=>i); // Subscribe to this to get flattened data var flattened = source.Switch() 

Running Publish().RefCount() creates a multicast source that subscribes once when the first subscription is done, and then unsubscribes when the last subscription is deleted. It is important to remember that only one source subscription is always made, which may or may not be important.