Camel Restlet异步发送对客户端的响应

Posted 2021-04-14

我有一个场景，我正在使用camel-restlet组件来接收发布请求，我将这些请求转发到外部Web服务，在收到外部服务的响应代码后，我需要将此响应代码添加到我自己的异步响应客户端。

我试图将响应对象保存到hashMap，其中key是基于请求内容生成的唯一序列号，一旦从外部Web服务接收到响应，我就可以使用此唯一键从hashMap检索响应对象。似乎像restlet保存对exchange.getOut（）消息的响应并同步发送回客户端，这不是我想要的。不设置out消息会给我一个nullPointerException。

路线类别：

public class ReceiveRoute extends RouteBuilder {

@Override
public void configure() throws Exception {

    from("restlet:http://localhost:8083/api/atmp?restletMethod=post")
        .to("activemq:queue:requestReceiveQueue");  

    from("activemq:queue:requestReceiveQueue")
        .process(new RequestProcessor())
        .to("activemq:queue:requestSendQueue");

    from("activemq:queue:requestSendQueue")
        .setHeader(Exchange.HTTP_METHOD, constant("POST"))
        .setHeader(Exchange.CONTENT_TYPE, constant("application/json"))
            .to("jetty:http://localhost:8080/rest_api_demo/api/restService")
            .bean("responseProcessor"); 
    }
}

requestProcessor类：

public class RequestProcessor implements Processor {

@Override
public void process(Exchange exchange) throws Exception {
    Message message = exchange.getIn();
    byte[] bytes = (byte[])message.getBody(); 
    String body = new String(bytes);

    String atmpId = GUIDGenerator.generateAtmpSerialNumber(); 
    String terIndentifier = GUIDGenerator.generateTerminalIdentifier(body);
    MapLookupHelper.insertResponse(atmpId, terIndentifier, exchange);

    Map<String, Object> messageMap = new HashMap<String, Object>();
    messageMap = FormatUtil.parseJson(body); 
    messageMap.put("ATMPId", atmpId);
    exchange.getIn().setBody(messageMap.toString());    
  }
}

responseProcessor类

@Component
public class ResponseProcessor implements Processor {

@Override
public void process(Exchange exchange) throws Exception {
    Message in = exchange.getIn();
    String responseCode = in.getHeader(Exchange.HTTP_RESPONSE_CODE).toString();
    String body = in.getBody().toString(); 
    Map<String, Object> resMap = new HashMap<String, Object>(); 

    resMap = FormatUtil.parseJson(body);
    String atmpId = resMap.get("ATMPId").toString();
    Exchange ex = MapLookupHelper.getOutMessage(atmpId);

    ex.getOut().setHeader("HostResponseCode", responseCode);
    ex.getOut().setBody(resMap.toString());
  }
}

我是Apache Camel的新手，想知道restlet是否是正确的方法，如果没有，有关如何在Camel中处理客户端异步响应的任何建议？ AsyncProcessor只能解决这种情况吗？

答案

我认为这不是restlet的问题。您的交换模式是InOut，这就是为什么所有jms-endpoint同步等待.bean（“responseProcessor”）的结果。即使您将模式更改为InOnly，您的客户端也不会异步接收响应。我认为你应该制作另一条路线的架构，如下所示：

from("restlet:http://localhost:8083/api/atmp_asyncRequest?restletMethod=post")
            .process(exchange -> {
                exchange.setProperty("uniqueRequestId", GUIDGenerator.generateAtmpSerialNumber());
            })
            .inOnly("seda:requestReceiveQueue")// here starts async processing of your request
            .process(exchange -> {
                exchange.getProperty("uniqueRequestId");
                // make here response for client with generated request id
            });

    from("seda:requestReceiveQueue")
            .process(exchange -> {
                // prepareprocess request if need
            })
            .setHeader(Exchange.HTTP_METHOD, constant("POST"))
            .setHeader(Exchange.CONTENT_TYPE, constant("application/json"))
            .to("jetty:http://localhost:8080/rest_api_demo/api/restService")
            .process(exchange -> {
                exchange.getProperty("uniqueRequestId");
                // save somewhere prepared response for client bound to generated request id
            });

    from("restlet:http://localhost:8083/api/atmp_getResponse?restletMethod=post")
            .process(exchange -> {
                String requestId = ;//extract request id from client's request
                Object body =  ;//find response that you saved asynchronously by extracted request id
                // if response not found, then async processing request not ended, so you should send message to client to continue polling
                exchange.getIn().setBody(body);
            });

如果你没有在客户端的异步响应回调服务器，这将工作。

此外，您可以使用Seda组件而不是jms，用于在路由之间排队任务。