使用ListenableFuture从JMS请求/应答中获得DeferredResult

Question

我一直在寻找机会来改进我们暴露给外部客户的Rest Api。在那次练习中，我发现我几乎没有利用我们的集成团队所做的决定，即使用JMS请求/应答而不是传统的阻塞SOAP请求/应答进行后端集成。

目前所有与消息代理的交互都是使用jmsOutboundGateway完成的，因此请求线程必须等待完成。为了扩展RestAPI，我想使用来自Spring MVC控制器的DeferredResult发送JMS回复。控制器与消息代理的交互如下所示:

                Controller --> GatewayProxy --> JMSOutboundGateway

我正在寻找机会使用ListenableFuture作为GatewayProxy的返回类型，但我无法找到使用spring集成实现它的适当方法。

下面是我从控制器调用的集成流程:

<int:gateway 
        service-interface="ae.emaratech.ngx.service.PermitSearchService" 
        default-request-channel="permit_search_input_channel" 
        default-reply-timeout="${broker.jms.gateway.min.consumers}"/>

    <int:channel id="permit_search_input_channel" />
    <int:chain input-channel="permit_search_input_channel">

    <int:header-enricher>
        <int:header name="person_number" expression="payload"/>
    </int:header-enricher>

    <int:transformer expression="#formatString(@api_messages['FIND_PERM_BY_PERSNO_MSG'],headers)"/>
    <int:header-filter header-names="JMS_*,jms_*,priority" pattern-match="true" />

    <int:header-enricher>
        <int:header name="jms_type" type="java.lang.String" value="1" overwrite="true"/>
    </int:header-enricher>

    <jms:outbound-gateway 
        request-destination="permitsInboundQueue"
        reply-destination="permitsOutboundQueue" 
        receive-timeout="${broker.jms.gateway.timeout}"
        correlation-key="Correlation_ID" 
        connection-factory="brokerConnectionFactory">
        <jms:reply-listener concurrent-consumers="${broker.jms.gateway.min.consumers}"  max-concurrent-consumers="${broker.jms.gateway.max.consumers}"/>
    </jms:outbound-gateway>

    <int-xml:xpath-filter throw-exception-on-rejection="true">
        <int-xml:xpath-expression expression="not(boolean(/*/ErrorDetails))"/>
    </int-xml:xpath-filter>             
    <int-xml:xslt-transformer
        xsl-resource="classpath:/META-INF/spring/integration/permit-to-json.xsl"
        result-type="StringResult" >
        </int-xml:xslt-transformer>
    <int:transformer expression="payload.toString()"/>

</int:chain> 

Answer 1

不知道你有什么问题，但功能看起来像:

ListenableFuture<String> result = this.asyncGateway.async("foo");
result.addCallback(new ListenableFutureCallback<String>() {
    @Override
    public void onSuccess(String result) {
        ...
    }
    @Override
    public void onFailure(Throwable t) {
        ...
    }
});

从4.1版开始可用