终止外部事件的订阅

Question

在应用程序中，我对外部HTTP端点使用长轮询。我使用Spring的反应式WebClient来做这件事。为了在应用程序停止时干净利落地关闭(并避免难看的Netty堆栈跟踪)，我将takeUntil()与一个EmitterProcessor实例一起使用，当Spring停止我的bean (我实现了SmartLifecycle)时，我在该实例上调用onNext()。

整个过程是这样的：

@Component
@RequiredArgsConstructor
@Slf4j
public class LongPollingMessageReceiver implements SmartLifecycle {
    private boolean running = true;

    private final EmitterProcessor<Boolean> shutdown = EmitterProcessor.create();

    private final BackendMessageReceiver backendMessageReceiver;

    public void waitForMessages() {
        Mono.defer(() -> backendMessageReceiver.receiveMessages()) // Calls WebClient
            .repeat()
            .takeUntilOther(shutdown)
            .subscribe(event -> {
                // do something when the http endpoint answers
            });
    }

    @Override
    public int getPhase() {
        // We need to cancel the subscriptions before Reactor/Netty shuts down.
        // Using @PreDestroy does not work because it is called *after* the Reactor/Netty shutdown.
        return 0;
    }

    @Override
    public void start() {
        // Not needed
    }

    @Override
    public void stop() {
        log.info("Stopping message subscriptions");
        shutdown.onNext(true);
        shutdown.onComplete();
        running = false;
    }

    @Override
    public boolean isRunning() {
        return running;
    }
}

就目前而言，整个机制运行良好。但是，EmitterProcessor被标记为@Deprecated，并且javadoc要求使用Sink。Sink不实现Publisher接口，因此无法传递到takeUntilOther()中。

我应该怎么做才能解决这个问题，而不会永远被困在小于3.5的项目中？

Answer 1

Sinks旨在作为以编程方式触发反应式事件的面向开发人员的应用编程接口。但是，如果没有一种方法将它们作为典型的Flux或Mono呈现给应用程序的其余部分，那么这就没有多大用处。

Sinks.Many有一个asFlux()视图来实现这一效果。类似地，Sinks.One和Sinks.Empty也有一个asMono()视图。

这就是你可以用来传递给takeUntilOther的东西。