Camel Restlet отправляет ответ клиенту асинхронно - PullRequest
Новая
       20

Camel Restlet отправляет ответ клиенту асинхронно

0 голосов
/ 30 августа 2018

У меня есть сценарий, который я использую компонент верблюжьего рестлета для получения почтовых запросов, я пересылаю эти запросы во внешний веб-сервис, после получения кода ответа от внешнего сервиса, мне нужно добавить этот код ответа на мой собственный ответ клиенту асинхронно.

Я пытаюсь сохранить объект ответа в hashMap, где ключ - это уникальный серийный номер, сгенерированный на основе содержимого запроса. После получения ответа от внешнего веб-сервиса я могу извлечь объект ответа из hashMap, используя этот уникальный ключ. , Похоже, что рестлет сохраняет ответ на сообщение exchange.getOut () и отправляет обратно клиенту синхронно, что мне не нужно. Отказ от установки сообщения даст мне исключение 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"); 
    }
}

запросПроцессор класса: 

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 class 

@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 и хотел бы знать, правильный ли путь рестлета, если нет, какие-либо предложения о том, как я могу обрабатывать асинхронные ответы для клиента в Camel? Является ли AsyncProcessor единственным решением для такого сценария?

1 Ответ

0 голосов
/ 31 августа 2018

Я думаю, что это не вопрос рестлета. Ваш шаблон обмена - InOut, поэтому все конечные точки jms ожидают синхронно результата вашего .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 -> {
                // prepare\process 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 для организации очередей между маршрутами.

...