エラー 'は、Spring統合アグリゲータ用の一方向の' MessageHandler 'です。DSL

Question

私は、DSLを使用してスプリング統合でいくつかのものをテストしようとしています。これは、これまでのところ唯一のテストで、流れは単純です：

• は別に集計

を記録し、それらを

を集約パラレル

でそれらをいくつかのメッセージに

プロセスを作成します（ログ）アグリゲータから、それは正常に動作しています：

@Bean 
public IntegrationFlow integrationFlow() { 
    return IntegrationFlows 
      .from(integerMessageSource(), c -> c.poller(Pollers.fixedRate(1, TimeUnit.SECONDS))) 
      .channel(MessageChannels.executor(Executors.newCachedThreadPool())) 
      .handle((GenericHandler<Integer>) (payload, headers) -> { 
       System.out.println("\t delaying message:" + payload + " on thread " 
         + Thread.currentThread().getName()); 
       try { 
        Thread.sleep(2000); 
       } catch (InterruptedException e) { 
        System.err.println(e.getMessage()); 
       } 
       return payload; 
      }) 
      .handle(this::logMessage) 
      .aggregate(a -> 
        a.releaseStrategy(g -> g.size()>10) 
        .outputProcessor(g -> 
          g.getMessages() 
            .stream() 
            .map(e -> e.getPayload().toString()) 
            .collect(Collectors.joining(","))) 

        ) 
      .handle(this::logMessage) 
      .get(); 

} 

。集約（..）の部分を除外すると、サンプルが動作しています。

小枝アグリゲータ、私は次の例外を取得：

Caused by: org.springframework.beans.factory.BeanCreationException: The 'currentComponent' (org.faboo.test.ParallelIntegrationApplication$$Lambda$9/1341404543@6fe1b4fb) is a one-way 'MessageHandler' and it isn't appropriate to configure 'outputChannel'. This is the end of the integration flow. 

は、私の知る限り理解し、それがアグリゲータからの出力がないと文句を言い？

完全なソースは、ここで見つけることができます：hithub

Answer 1

問題がアグリゲータ前handle()ある - 集約するものは何もありませんので、それは何の結果を生成しない...

 .handle(this::logMessage) 
     .aggregate(a -> 

おそらくlogMessage(Message<?>) Aありvoid戻り値の型。

アグリゲータの前にログを記録する場合は、wireTapを使用するか、logMessageを変更してログ後にMessage<?>に戻します。

 .wireTap(sf -> sf.handle(this::logMessage))