Приведем холодный пример:
Observable<Integer> cold = Observable.create(subscriber -> {
try {
for (int i = 0; i <= 42; i++) {
// avoid doing unnecessary work
if (!subscriber.isUnsubscribed()) {
break;
}
subscriber.onNext(i);
}
subscriber.onCompleted();
} catch (Throwable cause) {
subscriber.onError(cause);
}
});
он начинает выполняться с нуля для каждого нового подписчика:
// starts execution
cold.subscribe(...)
и может остановить выполнение, если подписчик откажется от подписки досрочно:
// stops execution
subscription.unsubscribe();
теперь, если вместо примера цикла for у нас есть какая-то фактическая бизнес-логика (которая не должна воспроизводиться для каждого подписчика, а должна выполняться в реальном времени), то мы имеем дело с горячим наблюдаемым ...
PublishSubject<Integer> hot = PublishSubject.create();
Thread thread = new Thread(() -> {
try {
for (int i = 0; i < 42; i++) {
// how to avoid unnecessary work when no one is subscribed?
hot.onNext(i);
}
hot.onCompleted();
} catch (Throwable cause) {
hot.onError(cause);
}
});
когда мы хотим, чтобы это началось, мы могли бы сделать
// stats work (although no one is subscribed)
thread.start();
отсюда первый вопрос: как начать работу только тогда, когда подписывается первый наблюдатель? (может быть, подключаемый наблюдаемый?)
и важный вопрос: как остановить работу, когда последний подписчик отписался? (Я не могу понять, как получить доступ к текущим подпискам для этого субъекта, и хотел бы найти чистое решение без общего глобального состояния, если такое решение существует )
Одно из решений, которое я могу придумать, - это поднять тему с помощью специального оператора, который будет управлять подписчиками ...