Почему .share () не влияет на холодные источники (autoconnect vs. refCount)?

Question

Flux<Integer> shared = Flux.just(1, 2).share();

shared.subscribe(System.out::println);
shared.subscribe(System.out::println);

Поскольку share() превращает поток в горячий, я ожидаю, что первый подписчик получит все значения, а второй - не получит ни одного, поскольку поток завершился во время подписки.Но вывод такой же, как и без share: 1 2 1 2, но он должен быть просто 1 2.

Когда я заменяю share() на publish.autoconnect(), он работает как положено.Почему это так?

Answer 1

Ответ прост, но мне потребовалось некоторое время, чтобы понять его.

share() - это сокращение для publish().refCount().refCount() похоже на autoConnect(), за исключением одной дополнительной функции: он отключается, когда все подписчики отменили или - и это здесь ситуация - поток завершен.

Первый shared.subscribe создает подписку (черезshare) к исходному потоку.Все значения отправляются немедленно, поток завершается, подписка отменяется.

Поскольку подписки сейчас нет, вторая shared.subscribe снова создает подписку, и поток снова начинается с начала.

autoConnect, однако, не отменяет подписку.Если вы используете его вместо refCount, подписка на исходный поток остается, но поскольку поток завершен, любой дополнительный подписчик не получит никаких значений.