Operator: Do

[benjchristensen]

http://msdn.microsoft.com/en-us/library/hh229804(v=vs.103).aspx
http://msdn.microsoft.com/en-us/library/hh229307(v=vs.103).aspx
http://msdn.microsoft.com/en-us/library/hh229659(v=vs.103).aspx
http://msdn.microsoft.com/en-us/library/hh229539(v=vs.103).aspx
http://msdn.microsoft.com/en-us/library/hh229830(v=vs.103).aspx

[prabirshrestha]

was trying to implement this. but since do is a java keyword we ca't have it as a method name. .net/js use Do with caps.
ReactiveCocoa uses doNext, doError and doCompleted. Should rxjava too use this naming standard?

[benjchristensen]

I don't see how we would have all 3 of doNext, doError and doCompleted as method names as those are the various overload arguments being passed in.

For the method name, I wish forEach wasn't used for the synchronous version as that is exactly what this is ...

Options I can think of are:

• doForEach
• doOnEach
• for
• each
• onEach
• doTo

This should be easy to implement using map since it seems to basically be map except it always returns the same type T instead of allowing it to be transformed. Am I understanding it correctly?

I believe the method signatures will end up looking like this ... once we figure out the name:

Observable<T> nameOfFunction(Observable<T> source, Action1<T> onNext)
Observable<T> nameOfFunction(Observable<T> source, Observer<T> observer)
Observable<T> nameOfFunction(Observable<T> source, Action1<T> onNext, Action onCompleted)
Observable<T> nameOfFunction(Observable<T> source, Action1<T> onNext, Action1<Exception> onError)
Observable<T> nameOfFunction(Observable<T> source, Action1<T> onNext, Action1<Exception> onError, Action onCompleted)

[benjchristensen]

Does this function allow modifying the values or does it always return the exact same Observable source? In other words, is this only for performing side-effects?

[prabirshrestha]

I think a function name with do prefix might be better. Coming from a .net rx/rac I would expect a do[...] method.

Yes it can be easily implemented using map but this would only work for onNext. Migt have to use mapMany or materialize/dematerialize combination for others.

Some info on rx do: http://stackoverflow.com/q/8732001/157260

[mva]

OperationMap creates a new MapObserver with every subscription. As a consequence, func is evaluated for every onNext as many times as there are subscriptions. While this might be ok for pure functions, it would be very inconvenient for a do observable that is only there for its side-effects, e.g. something that logs events.

[benjchristensen]

Generally operators in a sequence do execute each time they are subscribed to - unless something like publish, multicast, replay etc caches the results of a subscription and becomes the source of a new sequence.

The documentation I've read is very unclear regarding do but I don't see anything implying that it would ignore subsequence subscribe invocations.

For example:

def o = anObservable
      .take(5)
      .map({transformHere}}
      .merge(anotherObservable)
      .do({doSideEffectsHere})
// subscribe to it once
o.subscribe()
// then subscribe again with another transformation done
o.map({anotherTransformation}).subscribe()

With sequence o being subscribed to twice I don't see anything about it that says it should cache everything before it and only be invoked once. It seems to me that it will perform the side-effects twice unless the sequence has publish, multicast, refCount, etc put into:

def o = anObservable
      .take(5)
      .map({transformHere}}
      .merge(anotherObservable)
      .do({doSideEffectsHere})
      .publish().refCount()

The code above adds .publish().refCount() which will cause the previous operators to be executed only once and shared to all subsequent subscriptions (http://northhorizon.net/2011/sharing-in-rx/).

Here is the .Net code for do: https://rx.codeplex.com/SourceControl/changeset/view/332fa75ecdb1#Rx/NET/Source/System.Reactive.Linq/Reactive/Linq/Observable/Do.cs

I'm not experienced with C# so can't guarantee I understand it but I see nothing in that code that automatically is caching the sequence.

Thus, I think the do operator will be executed each time just like other operators when subscribed to.

Am I misunderstanding something about the contract specified by Rx.Net?

[benjchristensen]

@prabirshrestha It makes sense to keep the do* prefix, so what do you prefer of these?

• doForEach
• doOnEach
• doTo

Is there a better option?

I lean towards doOnEach because on fits the onNext\onError\onCompleted naming convention.

[prabirshrestha]

@benjchristensen Even I preferred doOnEach. That was what I started to use.

Thus, I think the do operator will be executed each time just like other operators when subscribed to.

Thats right. I will verify this with .net too.

@mva do operator can be used to follow divide and conquer pattern to.
Here is a c# sample code from http://www.codeproject.com/Articles/132966/Exploring-Reactive-Extensions-Rx-through-Twitter-a

Observable.FromEvent<TextChangedEventArgs>(searchTextBox, "TextChanged")
          .Select(e => ((TextBox)e.Sender).Text)
          .Where(text => text.Length > 2)
          .Do(s => searchResults.Opacity = 0.5)       // reduce list opacity when typing   
          .Throttle(TimeSpan.FromMilliseconds(400))
          .ObserveOnDispatcher()
          .Do(s => LoadingIndicator.Visibility = Visibility.Visible)  // show the loading indicator 
          .SelectMany(txt => searchTwitter(txt))
          .Select(searchRes => ParseTwitterSearch(searchRes))
          .ObserveOnDispatcher()
          .Do(s => LoadingIndicator.Visibility = Visibility.Collapsed) // hide the loading indicator
          .Do(s => searchResults.Opacity = 1)                          // return the list opacity to one
          .Subscribe(tweets => searchResults.ItemsSource = tweets);

I found using do in mobile apps this way really useful.

[mva]

@benjchristensen, I think I am starting to understand the distinction you are making. The side-effect of a do is not "owned" by the single observable instance, but rather by its potentially many subscribers.

Using logging as an example: this makes perfect sense for a cold/passive observable because every value send to a subscriber is independent of all other values, and deserves its own log entry.

On the other hand, if I want to log the output of a hot/active observable irregardless of the (possibly zero) number of subscribers, I would put the do on the path from the observable to a Publish/Connect. Here, exactly one log entry would be written for each value produced by the hot observable.

Thank you for your explanation!