У меня есть некоторые проблемы с пониманием того, как подписка / наблюдение работает в RxJava. Я создал простое приложение с наблюдаемым, которое испускает названия планет солнечной системы, выполняет некоторые картографические и фильтрационные действия и печатает результаты.

Как я понимаю, планирование работы в фоновом потоке выполняется черезsubscribeOn оператор (и, кажется, работает нормально).

Наблюдение за фоновым потоком также отлично работает сobserveOn оператор.

Но у меня проблемы с пониманием того, как наблюдать за вызывающим потоком (будь то основной поток или любой другой). Это легко сделать на Android сAndroidSchedulers.mainThread() оператор, но я не знаю, как добиться этого в чистом Java.

Вот мой код:

public class Main {

    public static void main(String[] args) throws InterruptedException {

        ThreadPoolExecutor executor = new ThreadPoolExecutor(3, 5, 3000L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<>());

        System.out.println("Main thread: " + getCurrentThreadInfo());

        Observable<String> stringObservable = Observable.from(Arrays.asList("Merkury", "Wenus", "Ziemia", "Mars", "Jowisz", "Saturn", "Uran", "Neptun", "Pluton"))
                .map(in -> {
                    System.out.println("map on: " + getCurrentThreadInfo());
                    return in.toUpperCase();
                })
                .filter(in -> {
                    System.out.println("filter on: " + getCurrentThreadInfo());
                    return in.contains("A");
                })
                .subscribeOn(Schedulers.from(executor));

        for (int i = 0; i < 5; i++) {
            Thread thread = new Thread("Thread-" + i) {
                @Override
                public void run() {
                    stringObservable
                            .buffer(5)
                            .subscribe(s -> System.out.println("Result " + s + " on: " + getCurrentThreadInfo()));
                }
            };
            thread.start();
        }

    }

    private static String getCurrentThreadInfo() {
        return Thread.currentThread().getName() + "(" + Thread.currentThread().getId() + ")";
    }
}

Наблюдается в созданной и подписывается на работу в одном из трех потоков от исполнителя. Это работает как ожидалось. Но как наблюдать результаты в динамически создаваемом потоке в цикле for? Есть ли способ создать планировщик из текущего потока?

Кроме того, я узнал, что после запуска этого кода он никогда не завершается, и я не знаю почему? :(