CountAndBlock.java

import com.google.common.collect.ImmutableList;
import io.reactivex.Observable;
import io.reactivex.schedulers.Schedulers;

import java.util.concurrent.ExecutionException;
import java.util.stream.Stream;

public class TestMain {

    public static void main(String[] args) throws ExecutionException, InterruptedException {
        Stream<Integer> ints = ImmutableList.of(1, 2, 3, 4, 5, 6)
                .stream().map(i -> {
                    System.out.println("emitting:" + i);
                    return i;
                });

        Long count = Observable.fromIterable(() -> ints.iterator())
                .observeOn(Schedulers.computation())
                .map(x -> x + 1)
                .observeOn(Schedulers.io())
                .doOnNext(System.out::println)
                .observeOn(Schedulers.computation())
                .count()
                .blockingGet();

        System.out.println("count: " + count);
    }
}

out.txt

emitting:1
emitting:2
emitting:3
emitting:4
2
emitting:5
emitting:6
3
4
5
6
7
count: 6

Process finished with exit code 0