Observable API Review

[benjchristensen]

We are almost done with 0.20 which is the last chance for deprecating and breaking things. Next release is 1.0 which locks down the public API.

Is there anything here we should delete or fix. This could include operators that we don't feel are trusted or fully baked. We can always add it back later after improving it, similar to how pivot was removed recently as it wasn't trusted.

Static Methods

amb(Iterable<? extends Observable<? extends T>>)
amb(Observable<? extends T>, Observable<? extends T>)
amb(Observable<? extends T>, Observable<? extends T>, Observable<? extends T>)
amb(Observable<? extends T>, Observable<? extends T>, Observable<? extends T>, Observable<? extends T>)
amb(Observable<? extends T>, Observable<? extends T>, Observable<? extends T>, Observable<? extends T>, Observable<? extends T>)
amb(Observable<? extends T>, Observable<? extends T>, Observable<? extends T>, Observable<? extends T>, Observable<? extends T>, Observable<? extends T>)
amb(Observable<? extends T>, Observable<? extends T>, Observable<? extends T>, Observable<? extends T>, Observable<? extends T>, Observable<? extends T>, Observable<? extends T>)
amb(Observable<? extends T>, Observable<? extends T>, Observable<? extends T>, Observable<? extends T>, Observable<? extends T>, Observable<? extends T>, Observable<? extends T>, Observable<? extends T>)
amb(Observable<? extends T>, Observable<? extends T>, Observable<? extends T>, Observable<? extends T>, Observable<? extends T>, Observable<? extends T>, Observable<? extends T>, Observable<? extends T>, Observable<? extends T>)
combineLatest(List<? extends Observable<? extends T>>, FuncN<? extends R>)
combineLatest(Observable<? extends T1>, Observable<? extends T2>, Func2<? super T1, ? super T2, ? extends R>)
combineLatest(Observable<? extends T1>, Observable<? extends T2>, Observable<? extends T3>, Func3<? super T1, ? super T2, ? super T3, ? extends R>)
combineLatest(Observable<? extends T1>, Observable<? extends T2>, Observable<? extends T3>, Observable<? extends T4>, Func4<? super T1, ? super T2, ? super T3, ? super T4, ? extends R>)
combineLatest(Observable<? extends T1>, Observable<? extends T2>, Observable<? extends T3>, Observable<? extends T4>, Observable<? extends T5>, Func5<? super T1, ? super T2, ? super T3, ? super T4, ? super T5, ? extends R>)
combineLatest(Observable<? extends T1>, Observable<? extends T2>, Observable<? extends T3>, Observable<? extends T4>, Observable<? extends T5>, Observable<? extends T6>, Func6<? super T1, ? super T2, ? super T3, ? super T4, ? super T5, ? super T6, ? extends R>)
combineLatest(Observable<? extends T1>, Observable<? extends T2>, Observable<? extends T3>, Observable<? extends T4>, Observable<? extends T5>, Observable<? extends T6>, Observable<? extends T7>, Func7<? super T1, ? super T2, ? super T3, ? super T4, ? super T5, ? super T6, ? super T7, ? extends R>)
combineLatest(Observable<? extends T1>, Observable<? extends T2>, Observable<? extends T3>, Observable<? extends T4>, Observable<? extends T5>, Observable<? extends T6>, Observable<? extends T7>, Observable<? extends T8>, Func8<? super T1, ? super T2, ? super T3, ? super T4, ? super T5, ? super T6, ? super T7, ? super T8, ? extends R>)
combineLatest(Observable<? extends T1>, Observable<? extends T2>, Observable<? extends T3>, Observable<? extends T4>, Observable<? extends T5>, Observable<? extends T6>, Observable<? extends T7>, Observable<? extends T8>, Observable<? extends T9>, Func9<? super T1, ? super T2, ? super T3, ? super T4, ? super T5, ? super T6, ? super T7, ? super T8, ? super T9, ? extends R>)
concat(Observable<? extends Observable<? extends T>>)
concat(Observable<? extends T>, Observable<? extends T>)
concat(Observable<? extends T>, Observable<? extends T>, Observable<? extends T>)
concat(Observable<? extends T>, Observable<? extends T>, Observable<? extends T>, Observable<? extends T>)
concat(Observable<? extends T>, Observable<? extends T>, Observable<? extends T>, Observable<? extends T>, Observable<? extends T>)
concat(Observable<? extends T>, Observable<? extends T>, Observable<? extends T>, Observable<? extends T>, Observable<? extends T>, Observable<? extends T>)
concat(Observable<? extends T>, Observable<? extends T>, Observable<? extends T>, Observable<? extends T>, Observable<? extends T>, Observable<? extends T>, Observable<? extends T>)
concat(Observable<? extends T>, Observable<? extends T>, Observable<? extends T>, Observable<? extends T>, Observable<? extends T>, Observable<? extends T>, Observable<? extends T>, Observable<? extends T>)
concat(Observable<? extends T>, Observable<? extends T>, Observable<? extends T>, Observable<? extends T>, Observable<? extends T>, Observable<? extends T>, Observable<? extends T>, Observable<? extends T>, Observable<? extends T>)
create(OnSubscribe<T>)
defer(Func0<Observable<T>>)
empty()
empty(Scheduler)
error(Throwable)
error(Throwable, Scheduler)
from(Future<? extends T>)
from(Future<? extends T>, long, TimeUnit)
from(Future<? extends T>, Scheduler)
from(Iterable<? extends T>)
from(T[])
interval(long, TimeUnit)
interval(long, TimeUnit, Scheduler)
merge(Iterable<? extends Observable<? extends T>>)
merge(Iterable<? extends Observable<? extends T>>, int)
merge(Iterable<? extends Observable<? extends T>>, int, Scheduler)
merge(Iterable<? extends Observable<? extends T>>, Scheduler)
merge(Observable<? extends Observable<? extends T>>)
merge(Observable<? extends Observable<? extends T>>, int)
merge(Observable<? extends T>, Observable<? extends T>)
merge(Observable<? extends T>, Observable<? extends T>, Observable<? extends T>)
merge(Observable<? extends T>, Observable<? extends T>, Observable<? extends T>, Observable<? extends T>)
merge(Observable<? extends T>, Observable<? extends T>, Observable<? extends T>, Observable<? extends T>, Observable<? extends T>)
merge(Observable<? extends T>, Observable<? extends T>, Observable<? extends T>, Observable<? extends T>, Observable<? extends T>, Observable<? extends T>)
merge(Observable<? extends T>, Observable<? extends T>, Observable<? extends T>, Observable<? extends T>, Observable<? extends T>, Observable<? extends T>, Observable<? extends T>)
merge(Observable<? extends T>, Observable<? extends T>, Observable<? extends T>, Observable<? extends T>, Observable<? extends T>, Observable<? extends T>, Observable<? extends T>, Observable<? extends T>)
merge(Observable<? extends T>, Observable<? extends T>, Observable<? extends T>, Observable<? extends T>, Observable<? extends T>, Observable<? extends T>, Observable<? extends T>, Observable<? extends T>, Observable<? extends T>)
merge(Observable<? extends T>[])
merge(Observable<? extends T>[], Scheduler)
mergeDelayError(Observable<? extends Observable<? extends T>>)
mergeDelayError(Observable<? extends T>, Observable<? extends T>)
mergeDelayError(Observable<? extends T>, Observable<? extends T>, Observable<? extends T>)
mergeDelayError(Observable<? extends T>, Observable<? extends T>, Observable<? extends T>, Observable<? extends T>)
mergeDelayError(Observable<? extends T>, Observable<? extends T>, Observable<? extends T>, Observable<? extends T>, Observable<? extends T>)
mergeDelayError(Observable<? extends T>, Observable<? extends T>, Observable<? extends T>, Observable<? extends T>, Observable<? extends T>, Observable<? extends T>)
mergeDelayError(Observable<? extends T>, Observable<? extends T>, Observable<? extends T>, Observable<? extends T>, Observable<? extends T>, Observable<? extends T>, Observable<? extends T>)
mergeDelayError(Observable<? extends T>, Observable<? extends T>, Observable<? extends T>, Observable<? extends T>, Observable<? extends T>, Observable<? extends T>, Observable<? extends T>, Observable<? extends T>)
mergeDelayError(Observable<? extends T>, Observable<? extends T>, Observable<? extends T>, Observable<? extends T>, Observable<? extends T>, Observable<? extends T>, Observable<? extends T>, Observable<? extends T>, Observable<? extends T>)
never()
parallelMerge(Observable<Observable<T>>, int)
parallelMerge(Observable<Observable<T>>, int, Scheduler)
range(int, int)
range(int, int, Scheduler)
sequenceEqual(Observable<? extends T>, Observable<? extends T>)
sequenceEqual(Observable<? extends T>, Observable<? extends T>, Func2<? super T, ? super T, Boolean>)
switchOnNext(Observable<? extends Observable<? extends T>>)
timer(long, long, TimeUnit)
timer(long, long, TimeUnit, Scheduler)
timer(long, TimeUnit)
timer(long, TimeUnit, Scheduler)
using(Func0<Resource>, Func1<Resource, ? extends Observable<? extends T>>)
zip(Iterable<? extends Observable<?>>, FuncN<? extends R>)
zip(Observable<? extends Observable<?>>, FuncN<? extends R>)
zip(Observable<? extends T1>, Observable<? extends T2>, Func2<? super T1, ? super T2, ? extends R>)
zip(Observable<? extends T1>, Observable<? extends T2>, Observable<? extends T3>, Func3<? super T1, ? super T2, ? super T3, ? extends R>)
zip(Observable<? extends T1>, Observable<? extends T2>, Observable<? extends T3>, Observable<? extends T4>, Func4<? super T1, ? super T2, ? super T3, ? super T4, ? extends R>)
zip(Observable<? extends T1>, Observable<? extends T2>, Observable<? extends T3>, Observable<? extends T4>, Observable<? extends T5>, Func5<? super T1, ? super T2, ? super T3, ? super T4, ? super T5, ? extends R>)
zip(Observable<? extends T1>, Observable<? extends T2>, Observable<? extends T3>, Observable<? extends T4>, Observable<? extends T5>, Observable<? extends T6>, Func6<? super T1, ? super T2, ? super T3, ? super T4, ? super T5, ? super T6, ? extends R>)
zip(Observable<? extends T1>, Observable<? extends T2>, Observable<? extends T3>, Observable<? extends T4>, Observable<? extends T5>, Observable<? extends T6>, Observable<? extends T7>, Func7<? super T1, ? super T2, ? super T3, ? super T4, ? super T5, ? super T6, ? super T7, ? extends R>)
zip(Observable<? extends T1>, Observable<? extends T2>, Observable<? extends T3>, Observable<? extends T4>, Observable<? extends T5>, Observable<? extends T6>, Observable<? extends T7>, Observable<? extends T8>, Func8<? super T1, ? super T2, ? super T3, ? super T4, ? super T5, ? super T6, ? super T7, ? super T8, ? extends R>)
zip(Observable<? extends T1>, Observable<? extends T2>, Observable<? extends T3>, Observable<? extends T4>, Observable<? extends T5>, Observable<? extends T6>, Observable<? extends T7>, Observable<? extends T8>, Observable<? extends T9>, Func9<? super T1, ? super T2, ? super T3, ? super T4, ? super T5, ? super T6, ? super T7, ? super T8, ? super T9, ? extends R>)

Instance Methods

all(Func1<? super T, Boolean>)
ambWith(Observable<? extends T>)
asObservable()
buffer(Func0<? extends Observable<? extends TClosing>>)
buffer(int)
buffer(int, int)
buffer(long, long, TimeUnit)
buffer(long, long, TimeUnit, Scheduler)
buffer(long, TimeUnit)
buffer(long, TimeUnit, int)
buffer(long, TimeUnit, int, Scheduler)
buffer(long, TimeUnit, Scheduler)
buffer(Observable<? extends TOpening>, Func1<? super TOpening, ? extends Observable<? extends TClosing>>)
buffer(Observable<B>)
buffer(Observable<B>, int)
cache()
cache(int)
cast(Class<R>)
collect(R, Action2<R, ? super T>)
compose(Transformer<T, R>)
concatMap(Func1<? super T, ? extends Observable<? extends R>>)
concatWith(Observable<? extends T>)
contains(Object)
count()
debounce(Func1<? super T, ? extends Observable<U>>)
debounce(long, TimeUnit)
debounce(long, TimeUnit, Scheduler)
defaultIfEmpty(T)
delay(Func0<? extends Observable<U>>, Func1<? super T, ? extends Observable<V>>)
delay(Func1<? super T, ? extends Observable<U>>)
delay(long, TimeUnit)
delay(long, TimeUnit, Scheduler)
delaySubscription(long, TimeUnit)
delaySubscription(long, TimeUnit, Scheduler)
dematerialize()
distinct()
distinct(Func1<? super T, ? extends U>)
distinctUntilChanged()
distinctUntilChanged(Func1<? super T, ? extends U>)
doOnCompleted(Action0)
doOnEach(Action1<Notification<? super T>>)
doOnEach(Observer<? super T>)
doOnError(Action1<Throwable>)
doOnNext(Action1<? super T>)
doOnTerminate(Action0)
elementAt(int)
elementAtOrDefault(int, T)
exists(Func1<? super T, Boolean>)
filter(Func1<? super T, Boolean>)
finallyDo(Action0)
first()
first(Func1<? super T, Boolean>)
firstOrDefault(T)
firstOrDefault(T, Func1<? super T, Boolean>)
flatMap(Func1<? super T, ? extends Observable<? extends R>>)
flatMap(Func1<? super T, ? extends Observable<? extends R>>, Func1<? super Throwable, ? extends Observable<? extends R>>, Func0<? extends Observable<? extends R>>)
flatMap(Func1<? super T, ? extends Observable<? extends U>>, Func2<? super T, ? super U, ? extends R>)
flatMapIterable(Func1<? super T, ? extends Iterable<? extends R>>)
flatMapIterable(Func1<? super T, ? extends Iterable<? extends U>>, Func2<? super T, ? super U, ? extends R>)
forEach(Action1<? super T>)
forEach(Action1<? super T>, Action1<Throwable>)
forEach(Action1<? super T>, Action1<Throwable>, Action0)
groupBy(Func1<? super T, ? extends K>)
groupBy(Func1<? super T, ? extends K>, Func1<? super T, ? extends R>)
groupByUntil(Func1<? super T, ? extends TKey>, Func1<? super GroupedObservable<TKey, T>, ? extends Observable<? extends TDuration>>)
groupByUntil(Func1<? super T, ? extends TKey>, Func1<? super T, ? extends TValue>, Func1<? super GroupedObservable<TKey, TValue>, ? extends Observable<? extends TDuration>>)
groupJoin(Observable<T2>, Func1<? super T, ? extends Observable<D1>>, Func1<? super T2, ? extends Observable<D2>>, Func2<? super T, ? super Observable<T2>, ? extends R>)
ignoreElements()
isEmpty()
join(Observable<TRight>, Func1<T, Observable<TLeftDuration>>, Func1<TRight, Observable<TRightDuration>>, Func2<T, TRight, R>)
last()
last(Func1<? super T, Boolean>)
lastOrDefault(T)
lastOrDefault(T, Func1<? super T, Boolean>)
lift(Operator<? extends R, ? super T>)
limit(int)
longCount()
map(Func1<? super T, ? extends R>)
materialize()
mergeWith(Observable<? extends T>)
multicast(Func0<? extends Subject<? super T, ? extends R>>)
multicast(Func0<? extends Subject<? super T, ? extends TIntermediate>>, Func1<? super Observable<TIntermediate>, ? extends Observable<TResult>>)
multicast(Subject<? super T, ? extends R>)
nest()
observeOn(Scheduler)
ofType(Class<R>)
onBackpressureBuffer()
onBackpressureDrop()
onErrorResumeNext(Func1<Throwable, ? extends Observable<? extends T>>)
onErrorResumeNext(Observable<? extends T>)
onErrorReturn(Func1<Throwable, ? extends T>)
onExceptionResumeNext(Observable<? extends T>)
parallel(Func1<Observable<T>, Observable<R>>)
parallel(Func1<Observable<T>, Observable<R>>, Scheduler)
publish()
publish(Func1<? super Observable<T>, ? extends Observable<R>>)
publish(Func1<? super Observable<T>, ? extends Observable<R>>, T)
publish(T)
publishLast()
publishLast(Func1<? super Observable<T>, ? extends Observable<R>>)
reduce(Func2<T, T, T>)
reduce(R, Func2<R, ? super T, R>)
repeat()
repeat(long)
repeat(long, Scheduler)
repeat(Scheduler)
repeatWhen(Func1<? super Observable<? extends Notification<?>>, ? extends Observable<? extends Notification<?>>>)
repeatWhen(Func1<? super Observable<? extends Notification<?>>, ? extends Observable<? extends Notification<?>>>, Scheduler)
replay()
replay(Func1<? super Observable<T>, ? extends Observable<R>>)
replay(Func1<? super Observable<T>, ? extends Observable<R>>, int)
replay(Func1<? super Observable<T>, ? extends Observable<R>>, int, long, TimeUnit)
replay(Func1<? super Observable<T>, ? extends Observable<R>>, int, long, TimeUnit, Scheduler)
replay(Func1<? super Observable<T>, ? extends Observable<R>>, int, Scheduler)
replay(Func1<? super Observable<T>, ? extends Observable<R>>, long, TimeUnit)
replay(Func1<? super Observable<T>, ? extends Observable<R>>, long, TimeUnit, Scheduler)
replay(Func1<? super Observable<T>, ? extends Observable<R>>, Scheduler)
replay(int)
replay(int, long, TimeUnit)
replay(int, long, TimeUnit, Scheduler)
replay(int, Scheduler)
replay(long, TimeUnit)
replay(long, TimeUnit, Scheduler)
replay(Scheduler)
retry()
retry(Func2<Integer, Throwable, Boolean>)
retry(long)
retryWhen(Func1<? super Observable<? extends Notification<?>>, ? extends Observable<? extends Notification<?>>>, Scheduler)
retryWhen(Func1<? super Observable<? extends Notification<?>>, ? extends Observable<?>>)
sample(long, TimeUnit)
sample(long, TimeUnit, Scheduler)
sample(Observable<U>)
scan(Func2<T, T, T>)
scan(R, Func2<R, ? super T, R>)
serialize()
share()
single()
single(Func1<? super T, Boolean>)
singleOrDefault(T)
singleOrDefault(T, Func1<? super T, Boolean>)
skip(int)
skip(long, TimeUnit)
skip(long, TimeUnit, Scheduler)
skipLast(int)
skipLast(long, TimeUnit)
skipLast(long, TimeUnit, Scheduler)
skipUntil(Observable<U>)
skipWhile(Func1<? super T, Boolean>)
skipWhileWithIndex(Func2<? super T, Integer, Boolean>)
startWith(Iterable<T>)
startWith(Iterable<T>, Scheduler)
startWith(Observable<T>)
startWith(T)
startWith(T, T)
startWith(T, T, T)
startWith(T, T, T, T)
startWith(T, T, T, T, T)
startWith(T, T, T, T, T, T)
startWith(T, T, T, T, T, T, T)
startWith(T, T, T, T, T, T, T, T)
startWith(T, T, T, T, T, T, T, T, T)
startWith(T[], Scheduler)
subscribe()
subscribe(Action1<? super T>)
subscribe(Action1<? super T>, Action1<Throwable>)
subscribe(Action1<? super T>, Action1<Throwable>, Action0)
subscribe(Observer<? super T>)
subscribe(Subscriber<? super T>)
subscribeOn(Scheduler)
switchMap(Func1<? super T, ? extends Observable<? extends R>>)
take(int)
take(long, TimeUnit)
take(long, TimeUnit, Scheduler)
takeFirst(Func1<? super T, Boolean>)
takeLast(int)
takeLast(int, long, TimeUnit)
takeLast(int, long, TimeUnit, Scheduler)
takeLast(long, TimeUnit)
takeLast(long, TimeUnit, Scheduler)
takeLastBuffer(int)
takeLastBuffer(int, long, TimeUnit)
takeLastBuffer(int, long, TimeUnit, Scheduler)
takeLastBuffer(long, TimeUnit)
takeLastBuffer(long, TimeUnit, Scheduler)
takeUntil(Observable<? extends E>)
takeWhile(Func1<? super T, Boolean>)
takeWhileWithIndex(Func2<? super T, ? super Integer, Boolean>)
throttleFirst(long, TimeUnit)
throttleFirst(long, TimeUnit, Scheduler)
throttleLast(long, TimeUnit)
throttleLast(long, TimeUnit, Scheduler)
throttleWithTimeout(long, TimeUnit)
throttleWithTimeout(long, TimeUnit, Scheduler)
timeInterval()
timeInterval(Scheduler)
timeout(Func0<? extends Observable<U>>, Func1<? super T, ? extends Observable<V>>)
timeout(Func0<? extends Observable<U>>, Func1<? super T, ? extends Observable<V>>, Observable<? extends T>)
timeout(Func1<? super T, ? extends Observable<V>>)
timeout(Func1<? super T, ? extends Observable<V>>, Observable<? extends T>)
timeout(long, TimeUnit)
timeout(long, TimeUnit, Observable<? extends T>)
timeout(long, TimeUnit, Observable<? extends T>, Scheduler)
timeout(long, TimeUnit, Scheduler)
timestamp()
timestamp(Scheduler)
toBlocking()
toList()
toMap(Func1<? super T, ? extends K>)
toMap(Func1<? super T, ? extends K>, Func1<? super T, ? extends V>)
toMap(Func1<? super T, ? extends K>, Func1<? super T, ? extends V>, Func0<? extends Map<K, V>>)
toMultimap(Func1<? super T, ? extends K>)
toMultimap(Func1<? super T, ? extends K>, Func1<? super T, ? extends V>)
toMultimap(Func1<? super T, ? extends K>, Func1<? super T, ? extends V>, Func0<? extends Map<K, Collection<V>>>)
toMultimap(Func1<? super T, ? extends K>, Func1<? super T, ? extends V>, Func0<? extends Map<K, Collection<V>>>, Func1<? super K, ? extends Collection<V>>)
toSortedList()
toSortedList(Func2<? super T, ? super T, Integer>)
unsafeSubscribe(Subscriber<? super T>)
unsubscribeOn(Scheduler)
window(Func0<? extends Observable<? extends TClosing>>)
window(int)
window(int, int)
window(long, long, TimeUnit)
window(long, long, TimeUnit, Scheduler)
window(long, TimeUnit)
window(long, TimeUnit, int)
window(long, TimeUnit, int, Scheduler)
window(long, TimeUnit, Scheduler)
window(Observable<? extends TOpening>, Func1<? super TOpening, ? extends Observable<? extends TClosing>>)
window(Observable<U>)
zipWith(Iterable<? extends T2>, Func2<? super T, ? super T2, ? extends R>)
zipWith(Observable<? extends T2>, Func2<? super T, ? super T2, ? extends R>)

[davidmoten]

I think Observable.using may have a design flaw as per #1466.

[dvtomas]

Hi all,
there is something in RxJava I don't have a clear understanding about, but still feel a little bit uneasy about. It is hard for me to distinguish for which (of my own Observables) to provide onError handlers and for which not. Basically, I have made the experience, that for some observables, it is normal (and in some cases even desirable) to end in an error state, but for others, it is either impossible, or it denotes a programming error. But the RxJava (RxScala) type system doesn't distinguish it in any way.

I'll give an example for a simplified version of my project. I have a measurement GUI written in RxScala that is connected via TCP/IP to a measurement server (C++). The GUI instructs the core to start a measurement, stop a measurement, and periodically requests measured data.

An example of an observable that may naturally end in an error state is an Observable of responses sent via TCP from the measurement server. It is created using Observable.using, and when the TCP connection fails, it is desirable for it to end in an error state. This way, the using operator will automatically allow me to clean up the TCP connection after failure.

Besides that, I have another Observable, one with an Event Bus semantics, currently implemented as a Subject (although I don't like that), that represents the state of the measurement - Disconnected from the core, Connecting, Connected, Measuring, ... and this Observable should never fail. It serves only as a basis for user feedback, and user should always have some feedback, even if it was about failed connection to the measurement server. If the user feedback fails, it means that the programmer (me) has made an error somewhere in the implementation. An onError handler doesn't make any sense, any such error should be propagated outsite the Observables chains as an ordinary RuntimeError.

For some observables, it is (almost) impossible to even get into an error state, like any operator treated with materialize

I really don't know how to do it, but I would like to see this distinction somehow encapsulated in the type system of RxJava, so that when I see an observable declaration in my code, I don't have to think (over and over again) whether it makes sense to provide an onError handler. I like the type system helping me wherever possible. If I filter an observable that cannot fail, I get another observable that cannot fail (unless I throw an exception in the predicate, but that should hopefully not be allowed). If I map an observable that cannot fail, I get an observable that can fail (or should we have two versions of map, one non-failing, the other possibly failing?) If I materialize an observable that can fail, I obtain an observable that never fails. When I create a Subject, I'd like to be given an opportunity to mark it as either non-failing, or with a possibility of failure. And so on...

Or, is it just me running into problems others don't encounter, possibly by my lame usage of RxJava?

Thank you for any thoughts on this, best regards,
Tomáš Dvořák

[abersnaze]

@dvtomas, @stealthcode and I have thought about this problem as a collection of traits that can be changed by operators. It ultimately falls apart because of closures passed in to operators like map. Suppose you had an Observable that doesn't onError and map was used, how would the closure signal that this not calling onError trait should kept the same, added or removed (not applicable to this operator and trait). The purest form of this problem is defer. We've haven't been able incorporate this in RxJava (core) in any reasonable way. I've been waiting until we start 2.0 to really start bugging @benjchristensen.

[dvtomas]

@abersnaze, As I said, I don't have any proposal on how to solve this problem. There could e.g. be two variants of the map operator (nonFailingMap and possiblyFailingMap, though the names would probably be different). For other operators, only a single variant is meaningful (filter preserves the failingness of the original, materialize always returns non-failing observable). I just wanted to mention this before v. 1.0, and the fact that Netflix has already given some thought to this is enough for me for the moment.

[benjchristensen]

@headinthebox is really the best one to way in on whether a type system can help here as he is the expert.

@abersnaze and @stealthcode have shared their experiments with me, but I don't think Java has a strong enough type system to do this.

I do agree @dvtomas that this is worth exploring further but that is belongs in 2.0 work. Similarly, I think it may be time to reopen the discussion whether we create an Rx Future so that Scalar and Vector use cases are represented by the types but work well together. This wasn't needed in Rx.Net because C# has a good Future in Task. On the JVM the j.u.c.Future isn't an option and CompletableFuture is not exactly a good mix as far as its APIs are concerned.

Let's pick this up in a couple months after the dust settles on 1.0. If you want, create a specific issue for it and I'll create a 2.0 milestone for tracking.

[jbripley]

retryWhen(Func1<? super Observable<? extends Notification<?>>, ? extends Observable<? extends Notification<?>>>, Scheduler)
retryWhen(Func1<? super Observable<? extends Notification<?>>, ? extends Observable<?>>)

Any particular reason why the signature of retryWhen, without an explicit Scheduler, has slightly differerent Func1 return value?

The repeatWhen methods, with or without and explicit Scheduler argument, do not differ in Func1 signature. So it would makes sense that retryWhen methods would match that or the Scheduler-less repeatWhen method is changed to match the Scheduler-less retryWhen.

[benjchristensen]

Any particular reason why the signature of retryWhen, without an explicit Scheduler, has slightly differerent Func1 return value?

@stealthcode is there a reason for this? It seems like they should be the same.

@jbripley good catch, I completely missed that.

[abersnaze]

@jbripley, @benjchristensen it's a mistake. Created PR.

[benjchristensen]

Thanks @abersnaze

[abersnaze]

are both doOnTerminate() and finallyDo() needed?

[benjchristensen]

They behave differently.

doOnTerminate happens-before onCompleted or onError whereas finallyDo happens-after.

        Observable.just("--event1").finallyDo(() -> {
            System.out.println("finally");
        }).map(s -> {
            System.out.println("mapping s: " + s);
            return s;
        }).forEach(System.out::println, e -> e.printStackTrace(), () -> System.out.println("Completed"));

        System.out.println("---------------------------------------");

        Observable.just("--event2").doOnTerminate(() -> {
            System.out.println("terminate");
        }).map(s -> {
            System.out.println("mapping s: " + s);
            return s;
        }).forEach(System.out::println, e -> e.printStackTrace(), () -> System.out.println("Completed"));

        System.out.println("---------------------------------------");

Output:

mapping s: --event1
--event1
Completed
finally
---------------------------------------
mapping s: --event2
--event2
terminate
Completed
---------------------------------------

Note how the "terminate` output is before "Completed" but "finally" is after.