将上下文从一个通量/单声道传递到另一个

Question

有趣的事情发生在webflux包内.但是,我在源头的旅程没有解决以下问题.

假设我有以下单声道(或声音):

Mono hello = Mono.empty()
            .subscriberContext(ctx -> ctx.put("message", "hello"));

我在webfilter中使用类似的构造来丰富具有租户和用户数据的管道.然后在控制器中使用这样的结构:

Mono world = Mono.subscriberContext()
            .map(ctx -> (String)ctx.get("message"));

hello mono的上下文填充在世界单声道中.我试图弄清楚这是如何完成的,也是为了进行单元测试.

最后它仍然是一个谜.我尝试使用mono/flux对象上的常规方法执行此操作,但是我没有成功地将hello上下文提供给world mono.如何融合助焊剂和单体并将上下文传递给上游操作员？

Answer 1

您想做几件事：

1.) 发布订阅者上下文

mono.subscriberContext({ Context context ->
    context.put("key", "value")
})

2.) 订阅/访问订阅者上下文

mono.flatMap({ def r ->
    return Mono.subscriberContext().map({ Context context ->
        context.get("key")
        context.get("keyOrMapOrStateObject").put("someKey", "someData")
        return r
    })
})

3.) 可能将数据从一个事件传递到下游事件

mono.flatMap({ def r ->
    return Mono.subscriberContext().map({ Context context ->
        def someData = context.get("keyOrMapOrStateObject").get("someKey")
        return r
    })
})

总的来说，它看起来像这样：（这是常规语法）

mono.flatMap({ def r ->
    return Mono.subscriberContext().map({ Context context ->
        context.get("key")
        context.get("keyOrMapOrStateObject").put("someKey", "someData")
        return r
    })
}).flatMap({ def r ->
    return Mono.subscriberContext().map({ Context context ->
        def someData = context.get("keyOrMapOrStateObject").get("someKey")
        return r
    })
}).subscriberContext({ Context context ->
    context.put("key", "value")
    context.put("keyOrMapOrStateObject", new HashMap())
})

这是一个粗略的轮廓 - 尚未按原样准备好，但它应该可以帮助您理解该模式。

祝你好运！

Answer 2

WebFlux 使用worldMono 并在其上构建反应链，以 reactor-netty 中的 HTTP 请求作为最终来源。该WebFilter是链建设的一部分，所以他们可以丰富Context整个链条。

IIRCMono.subscriberContext()将在一个被使用flatMap，这使得主序列Context提供给其内件，因此它可以看到Context的hello。