如何查看Spring集成的集成流中的类型

Question

我试图理解在Spring集成中聚合时返回的类型，这相当困难。我正在使用Project Reactor，我的代码片段是：

public FluxAggregatorMessageHandler randomIdsBatchAggregator() {
    FluxAggregatorMessageHandler f = new FluxAggregatorMessageHandler();
    f.setWindowTimespan(Duration.ofSeconds(5));
    f.setCombineFunction(messageFlux -> messageFlux
        .map(Message::getPayload)
        .collectList()
        .map(GenericMessage::new);
    return f;
}

@Bean
public IntegrationFlow dataPipeline() {
   return IntegrationFlows.from(somePublisher)
// ----> The type Message<?> passed? Or Flux<Message<?>>?
      .handle(randomIdsBatchAggregator())
// ----> What type has been returned from the aggregation?
      .handle(bla())
      .get();
}

除了理解示例中传递的类型之外，我还想知道如何才能知道IntegrationFlow中流动的对象及其类型。

Answer 1

IntegrationFlows.from(somePublisher)

这将在内部创建一个FluxMessageChannel，订阅提供的Publshier。每个事件都会从这个通道发送到它的订阅者——你的聚合器。

FluxAggregatorMessageHandler生成setCombineFunction（）JavaDocs中解释的内容：

/**
 * Configure a transformation {@link Function} to apply for a {@link Flux} window to emit.
 * Requires a {@link Mono} result with a {@link Message} as value as a combination result
 * of the incoming {@link Flux} for window.
 * By default a {@link Flux} for window is fully wrapped into a message with headers copied
 * from the first message in window. Such a {@link Flux} in the payload has to be subscribed
 * and consumed downstream.
 * @param combineFunction the {@link Function} to use for result windows transformation.
 */
public void setCombineFunction(Function<Flux<Message<?>>, Mono<Message<?>>> combineFunction) {

所以，这是一个带有消息的单声道，你可以用来实现。collectList（）。当框架从FluxAggregatorMessageHandler发出回复消息时，它订阅了Mono。因此你的。句柄（bla（））必须包含有效负载列表。这对于聚合器结果来说是很自然的。

在文档中查看更多：https://docs.spring.io/spring-integration/docs/current/reference/html/message-routing.html#flux-aggregator