java - Interval continue working after onErrorResumeNext -

i´m using interval operator, , want continue emitting items if exception happens on pipeline.

so try use onerrorresumenext emitting item in case of exception. seen after emitt item, interval stop emitting more items.

here unit test.

@test public void testintervalobservablewitherror() {     subscription subscription = observable.interval(50, timeunit.milliseconds)             .map(time -> "item\n")             .map(item -> item = null)             .map(string::tostring)             .onerrorresumenext(t-> observable.just("item error emitted"))             .subscribe(system.out::print, t->{                         system.out.println(t);                     }                    );     testsubscriber testsubscriber = new testsubscriber((observer) subscription);     testsubscriber.awaitterminalevent(20000, timeunit.milliseconds); } 

i´m confuse behaviour, why observable unsubscribe if it´s receiving item onerrorresumenext

solution:

after explanations, realize when error happens observable t´s complete. end wrapping observable can have exception observable , i´m using flatmap. main observable continue emitting items.

@test public void testintervalobservablewitherror() {     observable.interval(100, timeunit.milliseconds)             .map(time -> "item\n")             .flatmap(item -> observable.just(item)                     .map(string::tostring))             .subscribe(system.out::print);      testsubscriber testsubscriber = new testsubscriber();     testsubscriber.awaitterminalevent(5000, timeunit.milliseconds); } 

if there´s operator can magic know.

regrads

your subscription interrupts because when onerrorresumenext triggered, upstream finished error. , emit item instead of let exception goes downstream. in order keep upstream alive have prevent exception thrown on it.

for particular example solution can this:

...     .map(time -> "item\n")     .map(item -> item = null)     .map(item -> {         try {             return item.tostring();         } catch (nullpointerexception e) {             return "item error emitted";     })     //no onerrorresumenext()     .subscribe ... 

onerrorresumenext replaces error item , calls oncomplete.