在 doOnSubscribe、doOnSuccess 和 doOnError 中获取 Reactor 订阅者上下文

Question

我正在尝试实现 Reactor 订阅者上下文 (http://projectreactor.io/docs/core/release/reference/#context)，以便我可以将 SLF4J MDC 中的值传递到 Flux 中，然后我可以使用这些值进行日志记录。

我使用subscriberContext() 方法来设置如下值：

someFlux().subscriberContext(Context.of(MDC_ATTRIBUTE, MDC.get(MDC_ATTRIBUTE)));

我还可以访问链中的上下文。以平面地图为例：

.flatMap(r -> Mono.subscriberContext().map(ctx -> {
    String name = ctx.getOrDefault(MDC_ATTRIBUTE_NAME, "NO CTX");
    return r;
}))

doOnEach() 也有效：

.doOnEach(signal -> {
    Context ctx = signal.getContext();
    if (signal.isOnNext()) {
        try (MDC.MDCCloseable closeable = MDC.putCloseable(MDC_ATTRIBUTE_NAME, ctx.getOrDefault(MDC_ATTRIBUTE_NAME, "MAAAAN"))) {
            log.debug("FINISHED: {}", requestName);
        }
    }
})

只有一个问题。我想在doOnSubscribe、doOnError 和doOnSuccess 中记录一些内容。虽然我可以使用 doOnEach 来检查 signal.isOnNext() 或 signal.isOnComplete()，但我发现从未调用过 signal.isOnSubscribe()。

所以问题是：我怎样才能获得doOnSubscribe() 中的上下文，或者这根本不可能？

Answer 1

这是可能的，不是在 100% 的用例中，而且有一点技巧：

Flux.just("foo")
    .doOnSubscribe(sub -> {
        Scannable actual = Scannable.from(sub).scan(Scannable.Attr.ACTUAL);
        if (actual instanceof CoreSubscriber) {
            Context context = ((CoreSubscriber) actual).currentContext();

            System.out.println(context);
        }
    })
    .map(v -> "value: " + v) //below or above doOnSubscribe is fine
    .subscriberContext(Context.of("foo", "bar")) //MUST be below doOnSubscribe
    .blockLast();