Springインテグレーション "Publish Subscribe Channel" with Spring DSL

Question

私のフローが始まる単純なフローを構築しようとしていますが、HTTP受信リクエストをHTTP受信リクエストで受信し、それを 'SubscribableChannel'に公開します。このチャンネルを購読しているユーザーは「N」人になる可能性があります。下の図はフローを示しています。

私はこの流れを設定するには、春のDSLを使用しようとしていますし、それがうまくなってトラブルを抱えています。以下は私のコードです。

@Bean 
public IntegrationFlow receiveHttpPost() { 
    return IntegrationFlows.from(Http.inboundChannelAdapter("/receive") 
      .mappedRequestHeaders("*") 
      .requestChannel(httpInAdapterPubSubChannel())) 
      .transform(new ObjectToStringTransformer()) 
      .get(); 
} 

@Bean 
public SubscribableChannel httpInAdapterPubSubChannel() 
{ 
    return MessageChannels.publishSubscribe("httpInAdapterPubSubChannel") 
    .get(); 
} 

@Bean 
public IntegrationFlow subscriber1() { 
    return IntegrationFlows.from(httpInAdapterPubSubChannel()) 
      .handle(message -> System.out.println("Enrich Headers based on Payload....")) 
      .get(); 
} 

@Bean 
public IntegrationFlow subscriber2() { 
    return IntegrationFlows.from(httpInAdapterPubSubChannel()) 
       .handle(message -> System.out.println("Save Payload to Audit Table...")) 
       .get(); 
} 

私はこのフローを実行すると、私が手「のメッセージを処理するために失敗しました。ネストされた例外はorg.springframework.messaging.core.DestinationResolutionExceptionです：ノー出力チャンネルまたは利用可能replyChannelヘッダー」。

o.s.i.channel.PublishSubscribeChannel : preSend on channel 'httpInAdapterPubSubChannel', message: GenericMessage [payload=Test, headers={content-length=4, http_requestMethod=POST, accept-language=en-US,en;q=0.8, accept=*/*, host=localhost:8080, http_requestUrl=http://localhost:8080/receive, connection=keep-alive, content-type=text/plain;charset=UTF-8, id=2c6ee729-96ee-1ae5-be31-a9bc56092758, cache-control=no-cache, accept-encoding=gzip, deflate, br, user-agent=Mozilla/5.0 (Windows NT 10.0; WOW64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/55.0.2883.87 Safari/537.36, timestamp=1484457726393}] 
o.s.i.t.MessageTransformingHandler  : org.springframework.integration.transformer.MessageTransformingHandler#0 received message: GenericMessage [payload=Test, headers={content-length=4, http_requestMethod=POST, accept-language=en-US,en;q=0.8, accept=*/*, host=localhost:8080, http_requestUrl=http://localhost:8080/receive, connection=keep-alive, content-type=text/plain;charset=UTF-8, id=2c6ee729-96ee-1ae5-be31-a9bc56092758, cache-control=no-cache, accept-encoding=gzip, deflate, br, user-agent=Mozilla/5.0 (Windows NT 10.0; WOW64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/55.0.2883.87 Safari/537.36, timestamp=1484457726393}] 
o.a.c.c.C.[.[.[/].[dispatcherServlet] : Servlet.service() for servlet [dispatcherServlet] in context with path [] threw exception [Request processing failed; nested exception is org.springframework.messaging.MessagingException: Failed to handle Message; nested exception is org.springframework.messaging.core.DestinationResolutionException: no output-channel or replyChannel header available] with root cause 

org.springframework.messaging.core.DestinationResolutionException: no output-channel or replyChannel header available 
    at org.springframework.integration.handler.AbstractMessageProducingHandler.sendOutput(AbstractMessageProducingHandler.java:287) ~[spring-integration-core-4.3.6.RELEASE.jar:4.3.6.RELEASE] 
    at org.springframework.integration.handler.AbstractMessageProducingHandler.produceOutput(AbstractMessageProducingHandler.java:212) ~[spring-integration-core-4.3.6.RELEASE.jar:4.3.6.RELEASE] 
    at org.springframework.integration.handler.AbstractMessageProducingHandler.sendOutputs(AbstractMessageProducingHandler.java:129) ~[spring-integration-core-4.3.6.RELEASE.jar:4.3.6.RELEASE] 
    at org.springframework.integration.handler.AbstractReplyProducingMessageHandler.handleMessageInternal(AbstractReplyProducingMessageHandler.java:115) ~[spring-integration-core-4.3.6.RELEASE.jar:4.3.6.RELEASE] 
    at org.springframework.integration.handler.AbstractMessageHandler.handleMessage(AbstractMessageHandler.java:127) ~[spring-integration-core-4.3.6.RELEASE.jar:4.3.6.RELEASE] 
    at org.springframework.integration.dispatcher.BroadcastingDispatcher.invokeHandler(BroadcastingDispatcher.java:236) ~[spring-integration-core-4.3.6.RELEASE.jar:4.3.6.RELEASE] 
    at org.springframework.integration.dispatcher.BroadcastingDispatcher.dispatch(BroadcastingDispatcher.java:185) ~[spring-integration-core-4.3.6.RELEASE.jar:4.3.6.RELEASE] 
    at org.springframework.integration.channel.AbstractSubscribableChannel.doSend(AbstractSubscribableChannel.java:89) ~[spring-integration-core-4.3.6.RELEASE.jar:4.3.6.RELEASE] 
    at org.springframework.integration.channel.AbstractMessageChannel.send(AbstractMessageChannel.java:423) ~[spring-integration-core-4.3.6.RELEASE.jar:4.3.6.RELEASE] 
    at org.springframework.messaging.core.GenericMessagingTemplate.doSend(GenericMessagingTemplate.java:115) ~[spring-messaging-4.3.5.RELEASE.jar:4.3.5.RELEASE] 
    at org.springframework.messaging.core.GenericMessagingTemplate.doSend(GenericMessagingTemplate.java:45) ~[spring-messaging-4.3.5.RELEASE.jar:4.3.5.RELEASE] 

ここでは非常に間違ったことをしています。 「Publish Subscribe Channel」VIA Spring Integration DSLまたはJava設定を示す例を見つけようとしました。残念ながら、私は何も見つけることができませんでした： - /。誰かが私に例を提供し、私が持っている流れに間違っているものがあるかどうかを知るのを助けることができれば、心から感謝します。

「私が加入者1と加入者2を削除すると、私は同じエラーが表示されます。つまり、y HttpInboundAdapterを設定するときにやっていることが間違っています。

また、 'httpInAdapterPubSubChannel'をダイレクトに切り替えて、単一のルートフロー（分岐を使用しない）しかない場合は、正常に機能します。

Answer 1

.transform(new ObjectToStringTransformer())はどこかの結果を送信しようとしていますが、受信アダプターは応答を期待せず、トランスフォーマーはデータを送信できません。

おそらく、あなたはこのような何かを意味...

@Bean 
public IntegrationFlow receiveHttpPost() { 
    return IntegrationFlows.from(Http.inboundChannelAdapter("/receive") 
     .mappedRequestHeaders("*")) 
     .transform(new ObjectToStringTransformer()) 
     .channel(httpInAdapterPubSubChannel()) 
     .get(); 
} 

すなわちパブ/サブチャネルに変圧器の結果を送信します。

DSLリファレンスにはいくつかの例があります。例えば、hereおよびhereである。