如何在 Mule 中使用 ActiveMQ 将有效负载返回给调用者流？

Question

<jms:activemq-connector name="jms-connector"  brokerURL="${BrokerURL}" disableTemporaryReplyToDestinations="true"  maxRedelivery="10" durable="true" specification="1.1"/>
<jms:endpoint connector-ref="jms-connector"  name="SyncRequestsQueue" queue="SyncRequestsQueue">
    <jms:transaction action="ALWAYS_BEGIN"/>
</jms:endpoint>

<flow name = "RequestsEntryFlow" >
    <http:listener allowedMethods="POST"  parseRequest="false"  config-ref="HTTP_Request_Listener" path="/{name}" doc:name="HTTP Entry Flow" />
    <flow-ref name="requestBeanCreate" />
    <jms:outbound-endpoint connector-ref="jms-connector" exchange-pattern="request-response" ref="SyncRequestsQueue" /> 
</flow>

<flow name = "SyncRequestsQueueConsumer">
    <jms:inbound-endpoint  connector-ref="jms-connector" exchange-pattern="request-response" ref="SyncRequestsQueue" />
    <set-variable variableName="id" value="#[payload]" doc:name="Variable" />
    <set-variable variableName="destination" value="#[app.registry.routerBean.getDestination(app.registry.messageBean.getReceiverID(payload))]"  doc:name="Variable" />
    <set-payload value="#[app.registry.messageBean.sendRequestToDestination(payload)]" />
    <processor-chain>
        <http:outbound-endpoint address="http://#[flowVars.destination]" exchange-pattern="request-response">
             <transformer ref="contentTypeTextXML"/>
        </http:outbound-endpoint>                     
        <expression-component>
            app.registry.messageBean.sendResponseToSender(flowVars.id);
            app.registry.messageBean.messageProcessedSuccessfully(flowVars.id);
        </expression-component>
    </processor-chain>
</flow>

这是场景：

对远程服务的请求进入 RequestsEntryFlow。 我用它做一些日志记录并将其发送到 SyncRequestsQueue。 SyncRequestsQueueConsumer 使用队列并将内容发送到远程服务器。 现在我想将远程服务器的响应返回到 RequestsEntryFlow 并从那里返回到服务调用者。 但响应没有返回给服务调用者。

注意：当我从代码中删除 JMS 和 Activemq 并使用 SyncRequestsQueueConsumer 作为子流程时，一切正常。所以我猜 JMS 连接器没有按照我需要的方式工作。

代码有什么问题，我该如何解决？谢谢你

--编辑--

我已经更改了我的代码以采用范围，现在它看起来像这样：

<flow name = "RequestsEntryFlow" >
    <http:listener allowedMethods="POST"  parseRequest="false"  config-ref="HTTP_Request_Listener" path="/{name}" doc:name="HTTP Entry Flow" />
    <flow-ref name="requestBeanCreate" />                          

    <choice doc:name="Choice">
        <when expression="#[app.registry.messageBean.isEMCRequired(payload)]">
            <jms:outbound-endpoint connector-ref="jms-connector"  ref="EMCRequiredRequestsQueue" />
        </when>
        <otherwise>                         
         <request-reply>
                <jms:outbound-endpoint connector-ref="jms-connector" exchange-pattern="one-way" ref="SyncRequestsQueue" />
                <jms:inbound-endpoint connector-ref="jms-connector" exchange-pattern="one-way"  queue="ResponseTempQueue" >
                    <set-payload value="#[payload.get(0)]"/>
                    <remove-property propertyName="Content_Type"/>
                    <transformer ref="contentTypeTextXML"/>
                </jms:inbound-endpoint>
             </request-reply>
        </otherwise>
    </choice>        
</flow>

<flow name = "SyncRequestsQueueConsumer">
    <jms:inbound-endpoint connector-ref="jms-connector" exchange-pattern="request-response" ref="SyncRequestsQueue" />
    <set-variable variableName="id" value="#[payload]" doc:name="Variable" />
    <set-variable variableName="destination" value="#[app.registry.routerBean.getDestination(app.registry.messageBean.getReceiverID(payload))]"  doc:name="Variable" />
    <set-payload value="#[app.registry.messageBean.sendRequestToDestination(payload)]" />

    <http:outbound-endpoint address="http://#[flowVars.destination]" exchange-pattern="request-response">
    <transformer ref="contentTypeTextXML"/>
    </http:outbound-endpoint>                          
    <expression-component>
        app.registry.messageBean.sendResponseToSender(flowVars.id);
        app.registry.messageBean.messageProcessedSuccessfully(flowVars.id);
    </expression-component>

</flow>

如果我将电子记录器添加到我的代码中，它将打印正确的有效负载（来自服务器的响应）

                <jms:inbound-endpoint connector-ref="jms-connector" exchange-pattern="one-way"  queue="ResponseTempQueue" >
                    <set-payload value="#[payload.get(0)]"/>
                    <remove-property propertyName="Content_Type"/>
                    <transformer ref="contentTypeTextXML"/>
                    <logger level="INFO" message="#[payload]" />
                </jms:inbound-endpoint>

但是响应没有通过 HTTP 路由回发送者

Answer 1

尝试在 RequestsEntryFlow 中使用请求-回复范围

在请求块中的请求回复范围内定义您的 SyncRequestsQueue 和请求-回复范围的回复块中的 VM。

在流程末尾的 SyncRequestsQueueConsumer 流程​​中，定义与请求-回复范围的回复块中定义的队列名称相同的 VM。

Answer 2

我猜你没有正确使用请求-回复。您能否通过请求-回复范围的示例实现并了解请求-回复范围。

我希望这在一定程度上有所帮助：

<http:listener-config name="HTTP_Listener_Configuration" host="0.0.0.0" port="8081" doc:name="HTTP Listener Configuration"/>
    <flow name="raml-http-request-connectorFlow">
        <http:listener config-ref="HTTP_Listener_Configuration" path="/test" doc:name="HTTP"/>
        <set-payload value="heloo" doc:name="Set Payload"/>
        <request-reply doc:name="Request-Reply">
            <vm:outbound-endpoint exchange-pattern="one-way" path="input" doc:name="VM"/>
            <vm:inbound-endpoint exchange-pattern="one-way" path="output" doc:name="VM"/>
        </request-reply>
        <logger message="log payload is:::::::::::::::::::::#[payload]" level="INFO" doc:name="Logger"/>
    </flow>
    <flow name="raml-http-request-connectorFlow1">
        <vm:inbound-endpoint exchange-pattern="one-way" path="input" doc:name="VM"/>
        <set-payload value="#[payload+'  welome']" doc:name="Set Payload"/>
        <vm:outbound-endpoint exchange-pattern="one-way" path="output" doc:name="VM"/>
    </flow>