Реактор способ отмены подписки

Question

Я пытаюсь выяснить проект реактора, и я ищу способ отмены подписки. Я знаю, что после подписания, например, Flux, я могу получить ссылку на объект Cancellation, который можно использовать для отправки onCancel Signal, но это происходит только после подписки, и мне нужно сохранить эту ссылку в какой-то коллекции.

Есть ли лучший способ получить объект аннулирования? Или просто отменить Подписки. Возможно, какое-то место, которое содержит ссылку на все активные подписки - да, это будет потрясающе ...

Answer 1

В реакторе нет смысла отказываться от Subscription, прежде чем вы позвоните subscribe() (так как это очень метод, который создает Subscription и распространяет этот сигнал на цепочку, чтобы начать эмиссию данных).

Нет централизованного места со всеми подписками, что не имеет особого смысла, потому что вам нужно найти конкретные подписки, которые вы хотите отменить (и имейте в виду, что каждый оператор в вашей цепочке может использовать промежуточная подписка ...).

Обратите внимание, что некоторые операторы также отменит подписку от вашего имени! То есть в случае take(int), например, который отменит вверх достаточно один раз детали были выброшены:

Flux.just(1, 2, 3, 4).log().take(2).subscribe(System.out::println); 

Выведет:

14:17:48.729 [main] INFO reactor.Flux.Array.1 - | onSubscribe([Synchronous Fuseable] FluxArray.ArraySubscription) 
14:17:48.732 [main] INFO reactor.Flux.Array.1 - | request(unbounded) 
14:17:48.732 [main] INFO reactor.Flux.Array.1 - | onNext(1) 
1 
14:17:48.732 [main] INFO reactor.Flux.Array.1 - | onNext(2) 
2 
14:17:48.732 [main] INFO reactor.Flux.Array.1 - | cancel()