new method for ignoreElements then other observable?

[davidmoten]

I've bumped into this use case as few times where I want to ignore all elements of Observable<T> a and then return Observable<R> b. This is how I normally do it:

a.ignoreElements().castAs(R.class).concatWith(b);

If R is a generic type we are forced into a more verbose cast:

((Observable<R>)(Observable<?>) a.ignoreElements()).concatWith(b);

What I don't like about this is that it's hard to read and I wonder if a new method say Observable.ignoreElementsThen would be a good addition to the API. It's signature would be

public <R> Observable<R> ignoreElementsThen(Observable<R> observable);

The use case would then be expressed as

a.ignoreElementsThen(b);

Any support for this new method (or is there an idiom I'm missing for this use case)?

[abersnaze]

@NiteshKant and I were working on something extremely similiar for RxNetty. With @benjchristensen we agreed to the names of the instance methods concatEmptyWith(Observable<R>) and mergeEmptyWith(Observable<R>). The only difference is that ours doesn't ignore the elements implicitly, but will onError if the source emits any values. For your use case it you would still use it with ignoreElements(). I might be able to be convinced that it should ignore the onNext or have a parameter to configure what to do. Check out #3060.

[davidmoten]

Thanks @abersnaze, so my use case would be expressed as

a.ignoreElements().concatEmptyWith(b);

That helps definitely. I'd go the extra yard though and add ignoreElementsThen just for simplicity and discoverability purposes. This is a recurring question for newbies:

http://stackoverflow.com/questions/31732054/rxjava-start-one-observable-when-other-one-finish

http://stackoverflow.com/questions/31616874/how-to-execute-2-observables-in-parallel-ignoring-their-results-and-execute-nex

http://stackoverflow.com/questions/31622332/using-takewhilewithindex-to-iterate-over-array-of-index-and-then-emit-an-observa

a.ignoreElementsThen(b);

[stealthcode]

I think I might have something to add to this discussion. I frequently use the following combination. Does anyone else see this issue as a common pattern?

Observable<T> inputs;
inputs.publish(is -> 
   potentiallyFailingWork(is)
      .doOnError(this::logRetryableFailure)
      .retryWhen(this::tryRetry)
      .doOnError(this::logFailure)
      .cache()
      .ignoreElements()
      .map(x -> (T) null)
      .concatEager(is)
)
.doOnNext(this::someOtherWork)
.retryWhen(this::tryRetry)
.cache();

The take away here is the sporadic use of cache(). This is to prevent later retry or resubscribes from saved observable references. I see this as preventing 2 things from happening in an observable chain. Firstly, if there is an error encountered downstream the original producer will not receive an unsubscribe() so it can continue to stream. Secondly the resubscribes are prevented because the data is simply cached and served to all subsequent subscribers. Is there a better pattern I could be using here?

[akarnokd]

Looks like the referenced PRs were closed in favor of Completable, which has andThen methods. Since this issue is a bit old, I'm closing this for now. Let us know if you have further input or proposition in this matter.