RxJava: calling unsubscribe from within onNext

Question

I wonder whether it is legal to call unsubscribe from within onNext handler like that:

List<Integer> gatheredItems = new ArrayList<>();

Subscriber<Integer> subscriber = new Subscriber<Integer>() {
    public void onNext(Integer item) {
        gatheredItems.add(item);
        if (item == 3) {
            unsubscribe();
        }
    }
    public void onCompleted() {
        // noop
    }
    public void onError(Throwable sourceError) {
        // noop
    }
};

Observable<Integer> source = Observable.range(0,100);

source.subscribe(subscriber);
sleep(1000);
System.out.println(gatheredItems);

The above code correctly outputs that just four elements were gathered: [0, 1, 2, 3]. But if somebody changes the source observable to be cached:

Observable<Integer> source = Observable.range(0,100).cache();

Then all hundred elements are gathered. I don't have a control over source observable (whether it is cached or not), so how to definitely unsubscribe from within onNext?

BTW: So is unsubscribing within onNext wrong thing to do?

(My actual use case is that in onNext I'm actually writing to output stream, and when an IOException occurs nothing more could be written to the output so I need to somehow stop further processing.)

Answer 1

The cache() caches everything once the first subscriber arrives and it doesn't give any options to stop it from downstream. You need to stop the stream before the cache() to avoid too much retention:

Observable<Integer> source = Observable.range(1, 100);

PublishSubject<Integer> stop = PublishSubject.create();

source
.doOnNext(v -> System.out.println("Generating " + v))
.takeUntil(stop)
.cache()
.doOnNext(new Action1<Integer>() {
    int calls;
    @Override
    public void call(Integer t) {
        System.out.println("Saving " + t);
        if (++calls == 3) {
            stop.onNext(1);
        }
    }
})
.subscribe();

Edit: the example above doesn't work below 1.0.13 so here is a version which should:

SerialSubscription ssub = new SerialSubscription();

ConnectableObservable<Integer> co = source
        .doOnNext(v -> System.out.println("Generating " + v))
        .replay();

co.doOnNext(v -> {
    System.out.println("Saving " + v);
    if (v == 3) {
        ssub.unsubscribe();
    }
})
.subscribe();

co.connect(v -> ssub.set(v));