RxJava: subscribe unsubscribe () method is not called

Question

In the code below, I need to free some resources when unsubscribing (where it registers "release").

   Observable first = Observable.create(new Observable.OnSubscribe<Object>() {
                    @Override
                    public void call(Subscriber<? super Object> subscriber) {
                        subscriber.add(Subscriptions.create(() -> {
                            log("release");         
                        }));
                    }
                }).doOnUnsubscribe(() -> log("first"));
   Observable second = Observable.create(…).doOnUnsubscribe(() -> log("second"));
   Observable result = first.mergeWith(second).doOnUnsubscribe(() -> log("result"));
   Subscription subscription = result.subscribe(…);
   //…
   subscription.unsubscribe();

But it only records the “result”. The forbidden subscription does not seem to apply to merged child observables. So, how to handle unsubscribing inside the first observable Observable.OnSubscribe?

Answer 1

unsubscribe , : , . , , , , onError onCompleted, 1.x.

, , using, :

Observable.using(
    () -> "resource", 
    r -> Observable.just(r), 
    r -> System.out.println("Releasing " + r))
.subscribe(System.out::println);