API Design Review: Replay Selector T -> R without ConnectableObservable

[benjchristensen]

Why are there replay overloads that:

• convert from T to R like a map operator?
• return Observable instead of ConnectableObservable?

public <R> Observable<R> replay(Func1<? super Observable<T>, ? extends Observable<R>> selector)

It seems that someone should just use observable.map(T -> R).replay() and that it should always return ConnectableObservable.

/cc @headinthebox and @jhusain

[akarnokd]

I never understood this operator, and the current implementation works different than Rx.NET.

public class ReplayMulticast {
    public static void main(String[] args) {
        Observable<Integer> source = Observable.from(1, 2, 3, 4)
                .doOnNext(v -> { 
                    System.out.println("Sideeffect"); 
                });

        Observable<Integer> result = source.replay(o -> o.take(2));

        for (int i = 1; i < 3; i++) {
            System.out.printf("- %d -%n", i);
            result.subscribe(System.out::println, Throwable::printStackTrace, 
                    () -> System.out.println("Done")
            );
        }
    }
}

that prints

- 1 -
Sideeffect
1
Sideeffect
2
Done
Sideeffect
Sideeffect
- 2 -
Sideeffect
1
Sideeffect
2
Done
Sideeffect
Sideeffect

An infinite stream would never stop.

Rx.NET prints

- 1 -
Sideeffect
1
Sideeffect
2
Done
- 2 -
Sideeffect
1
Sideeffect
2
Done

Regardless, I don't get the operator.

[headinthebox]

Replay has way to many overloads ... (personally I never use replay, so I am not the right person to ask which ones to keep :-)

[benjchristensen]

@headinthebox Now is the time for a review of these and deleting any if we want to.

[benjchristensen]

Can you specify which ones so we don't confuse each other on subjective non-obvious decisions?

[headinthebox]

Here are all the overloads.

The ones that create a ConnectableObservable:

ConnectableObservable<T> replay()
ConnectableObservable<T> replay(int bufferSize) 
ConnectableObservable<T> replay(int bufferSize, long time, TimeUnit unit)
ConnectableObservable<T> replay(int bufferSize, long time, TimeUnit unit, Scheduler scheduler)
ConnectableObservable<T> replay(int bufferSize, Scheduler scheduler)
ConnectableObservable<T> replay(long time, TimeUnit unit)
ConnectableObservable<T> replay(long time, TimeUnit unit, Scheduler scheduler)
ConnectableObservable<T> replay(Scheduler scheduler)

And the corresponding ones that take a function and create a regular observable:

Observable<R> replay(Func1<? super Observable<T>, ? extends Observable<R>> selector)
Observable<R> replay(Func1<? super Observable<T>, ? extends Observable<R>> selector, final int bufferSize) 
Observable<R> replay(Func1<? super Observable<T>, ? extends Observable<R>> selector, int bufferSize, long time, TimeUnit unit)
Observable<R> replay(Func1<? super Observable<T>, ? extends Observable<R>> selector, final int bufferSize, final long time, final TimeUnit unit, final Scheduler scheduler)
Observable<R> replay(Func1<? super Observable<T>, ? extends Observable<R>> selector, final int bufferSize, final Scheduler scheduler) 
Observable<R> replay(Func1<? super Observable<T>, ? extends Observable<R>> selector, long time, TimeUnit unit) 
Observable<R> replay(Func1<? super Observable<T>, ? extends Observable<R>> selector, final long time, final TimeUnit unit, final Scheduler scheduler)
Observable<R> replay(Func1<? super Observable<T>, ? extends Observable<R>> selector, final Scheduler scheduler)

These latter ones are similar to the Publish overload that takes a function Observable<R> publish(Func1<? super Observable<T>, ? extends Observable<R>> selector) where the source observable is published inside the function.

If you like replay (which I personally don't, in fact I have not used it in the last 2 years) all the overloads make sense. So, unfortunately, I guess that if we keep replay, we need all of them.

[headinthebox]

@akarnokd The .NET behavior is the correct one. When you call source.replay(replayedSource -> replayedSource.take(2)) the function gets a private copy source.replay() that it can use. Since source is finite, it should replay the side effects and take the first two elements.

[akarnokd]

@headinthebox thanks, I suspected so. Luckily, the issue I mentioned above was fixed and RxJava should behave as Rx.NET now.

[headinthebox]

@akarnokd great! and @benjchristensen sorry for not being able to prune this bunch, but it makes Rx extra super :-)

[benjchristensen]

Closing out ... review has decided that what we have is what we want.

Thanks @akarnokd and @headinthebox