flatMapしてるものは、unsubscribeしたら実行されないよね確認

rxjava_unsub_flatmap_test.java

import rx.Observable;
import rx.Single;
import rx.Subscription;

import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;

public class Hoge {
    public static void main(String[] args) {

        CountDownLatch countDownLatch = new CountDownLatch(2);

        SingleGenerator generator = new SingleGenerator(countDownLatch);

        Subscription subscription = generator.generate()
                .flatMap(_val -> generator.generate())
                .flatMap(_val -> generator.generate())
                .flatMap(_val -> generator.generate())
                .subscribe();

        try {
            countDownLatch.await();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }

        System.out.println("-- unsub --");
        subscription.unsubscribe();

        try {
            Thread.sleep(5000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        System.out.println("== exit ==");
    }

    private static class SingleGenerator {
        private int i;
        private final CountDownLatch countDownLatch;
        public SingleGenerator(CountDownLatch countDownLatch) {
            i = 0;
            this.countDownLatch = countDownLatch;
        }

        private Single<Integer> generate() {
            return Observable.timer(2, TimeUnit.SECONDS)
                    .toSingle()
                    .flatMap(_val -> Single.just(i++))
                    .doOnSuccess(val -> {
                        System.out.println(val);
                        countDownLatch.countDown();
                    });
        }
    }
}

ubsubした場合

0
1
-- unsub --
== exit ==

ubsubしなかった場合

0
1
-- unsub --
2
3
== exit ==