Java 类org.apache.camel.Exchange 实例源码

@Test
public void multipleCities() throws Exception {
    mockGlobalWeatherResponseBody = "<NewDataSet><Table><Country>TEST</Country><City>AA</City></Table><Table><Country>TEST</Country><City>BB</City></Table></NewDataSet>";

    Exchange response =  fluentProducerTemplate.send();
    String responseBody = response.getIn().getBody(String.class);
    CitiesResponse citiesResponse = objectMapper.readValue(responseBody,CitiesResponse.class);

    assertEquals(200, response.getIn().getHeader(Exchange.HTTP_RESPONSE_CODE));
    assertEquals("TEST", citiesResponse.getCountry());

    List<City> expectedCities = Arrays.asList(
            new City("AA", Arrays.asList("ZIP-AA")),
            new City("BB", Arrays.asList("ZIP-BB1","ZIP-BB2"))
    );
    //Sort response list by city name
    citiesResponse.getCities().sort(Comparator.comparing(City::getName));
    assertEquals(expectedCities, citiesResponse.getCities());
}

public void process(Exchange exchange) throws Exception {
    String key = (String)exchange.getIn().getHeader(WebsocketConstants.CONNECTION_KEY);

    MDC.clear();
    MDC.put("WebsocketConstants.CONNECTION_KEY",key);

    logger.info("Headers: {}",exchange.getIn().getHeaders());
}

@Override
public void process(Exchange exchange) throws Exception {
    String header = (String) exchange.getIn().getHeader("Authorization");

    if (header == null) {
        throw new UnauthorizedException();
    }

    Set<ITokenValidation> providers =
            exchange.getContext().getRegistry().findByType(ITokenValidation.class);

    if (providers.isEmpty()) {
        throw new UnauthorizedException();
    }

    ITokenValidation tokenProvider = providers.iterator().next();

    String token = header.replace("TOKEN", "");
    token = token.trim();
    boolean isvalid = tokenProvider.isTokenvalid(token);

    if (!isvalid) {
        throw new UnauthorizedException();
    }
}

@Override
public void process(Exchange exchange) throws Exception {
    DiskFileItemFactory factory = new DiskFileItemFactory();
    factory.setRepository(new File(System.getProperty("java.io.tmpdir")));

    ExchangeFileUpload upload = new ExchangeFileUpload(factory);

    java.util.List<FileItem> items = upload.parseExchange(exchange);

    if(items.size() >= 1){
        exchange.getIn().setBody(items.get(0).getInputStream());

        for (int i = 1; i < items.size(); i++) {
            exchange.setProperty(items.get(i).getName(), items.get(i).getInputStream());
        }
    }
}

@FallbackConverter
public static <T> T convertTo(Class<T> type, Exchange exchange, Object value, TypeConverterRegistry registry) {

    try {
        if (value != null && value.getClass().equals(String.class)) {
            if (value.toString().startsWith("{") || value.toString().startsWith("[")) {
                T result = new UnirestJacksonObjectMapper().readValue(value.toString(), type);

                return result;
            }
        }
    }
    catch(Exception ex){
        return null;
    }
    return null;
}

@Test
public void testSDPParseOffer() throws Exception {
    Exchange ex = new DefaultExchange(new DefaultCamelContext());
    ex.getIn().setBody(offerSdp);
    processor.process(ex);
    assertEquals(ex.getIn().getBody().getClass(),Offer.class);

    Offer offer = (Offer)ex.getIn().getBody();

    MediaDescription mediaDescription = (MediaDescription) offer.getSdp().getMediaDescriptions(true).get(0);

    String icePwd = mediaDescription.getAttribute("ice-pwd");
    String iceUfrag = mediaDescription.getAttribute("ice-ufrag");
    String fingerprint = offer.getSdp().getAttribute("fingerprint");

    assertEquals(icePwd,"c490fef46f74bdbe64edd636bc49a259");
    assertEquals(iceUfrag,"64dc2277");
    assertEquals(fingerprint,"sha-256 99:45:B1:94:7E:97:AE:F2:A5:75:86:89:B5:AD:06:BB:63:02:FA:05:04:B2:83:1B:52:C9:EF:0E:61:8F:38:73");
}

@Override
public Exchange aggregate(Exchange oldExchange, Exchange newExchange) {
    Exchange exchange = super.aggregate(oldExchange, newExchange);

    List<Exchange> aggregatedExchanges = exchange.getProperty("CamelGroupedExchange", List.class);

    // Complete aggregation if we have "numberOfSystems" (currently 3) different messages (where "system" headers are different)
    // https://github.com/apache/camel/blob/master/camel-core/src/main/docs/eips/aggregate-eip.adoc#completing-current-group-decided-from-the-aggregationstrategy
    if (numberOfSystems == aggregatedExchanges.stream().map(e -> e.getIn().getHeader("system", String.class)).distinct().count()) {
        exchange.setProperty(Exchange.AGGREGATION_COMPLETE_CURRENT_GROUP, true);
    }

    return exchange;
}

@Override
public final void process(final Exchange exchange) throws Exception {
    if (exchange.isFailed()) {
        return;
    }

    if (type == null) {
        return;
    }

    final Message message = message(exchange);
    final String bodyAsString = message.getBody(String.class);

    if (bodyAsString == null) {
        return;
    }

    try {
        final Object output = MAPPER.readValue(bodyAsString, type);
        message.setBody(output);
    } catch (final IOException e) {
        exchange.setException(e);
    }
}

@Override
public void process(Exchange exchange) throws Exception {


    if (exchange.getIn().getHeader("X-Request-ID") == null || exchange.getIn().getHeader("X-Request-ID").toString().isEmpty()) {
        exchange.getIn().removeHeader("X-Request-ID");
    }

    if (exchange.getIn().getHeader("X-Forwarded-For") == null || exchange.getIn().getHeader("X-Forwarded-For").toString().isEmpty()) {
        exchange.getIn().removeHeader("X-Forwarded-For");
    }

    if (exchange.getIn().getHeader("X-Forwarded-Host") == null || exchange.getIn().getHeader("X-Forwarded-Host").toString().isEmpty()) {
        exchange.getIn().removeHeader("X-Forwarded-Host");
    }

}

private Iterable evaluateSplitRule(Exchange origExchange) {
    final Object splitResult = splitRule.evaluate(origExchange, Object.class);

    if (null == splitResult) {
        return Collections.emptySet();
    }

    if (splitResult instanceof Iterable) {
        return (Iterable) splitResult;
    }

    if (splitResult instanceof Iterator) {
        return () -> (Iterator) splitResult;
    }

    if (splitResult.getClass().isArray()) {
        return Arrays.asList((Object[]) splitResult);
    }

    return Collections.singleton(splitResult);
}

@Test
public void testReplayUnhandled() throws InterruptedException {
    mock.expectedMessageCount(0);
    error.expectedMessageCount(0);
    Exchange result = producerTemplate.send("direct:flow-test-5", createExchange("test"));
    mock.assertIsSatisfied();
    error.assertIsSatisfied();
    Long flowId = flowId(result); 
    mock.reset();
    error.reset();
    mock.expectedMessageCount(0);
    error.expectedMessageCount(0);
    flowManager.replayFlow(flowId);
    mock.assertIsSatisfied();
    error.assertIsSatisfied();
}

@Override
public void configure() throws Exception {

    from("sql-stored-start-connector:DEMO_OUT( "
            + "OUT INTEGER C )?schedulerPeriod=5000")
    .process(new Processor() {

        public void process(Exchange exchange)
                throws Exception {
            System.out.println(exchange.getIn()
                    .getBody().getClass());
            System.out.println(exchange.getIn()
                    .getBody());
        }
    });
}

@Test
public void noZipCity() throws Exception {
    mockGlobalWeatherResponseBody = "<NewDataSet><Table><Country>TEST</Country><City>XX</City></Table></NewDataSet>";

       Exchange response =  fluentProducerTemplate.send();
       String responseBody = response.getIn().getBody(String.class);
       CitiesResponse citiesResponse = objectMapper.readValue(responseBody,CitiesResponse.class);

       assertEquals(200, response.getIn().getHeader(Exchange.HTTP_RESPONSE_CODE));
       assertEquals("TEST", citiesResponse.getCountry());

       List<City> expectedCities = Arrays.asList(
               new City("XX", Arrays.asList())
       );
       assertEquals(expectedCities, citiesResponse.getCities());
}

/**
 * Creates a {@link PlatformPacket} from an <code>exchange</code> applying
 * either native type conversion to byte array or using
 * {@link #inFormat(DataFormat)} (higher priority). If
 * {@link #outConversion} is <code>false</code> no further conversion of
 * the body of the {@link Exchange#getIn()} message is done, otherwise,
 * {@link #outFormat} or {@link #outType} are applied. If both are not set a
 * conversion to {@link InputStream} type is made.
 * 
 * @param exchange message exchange.
 */
@Override
public PlatformPacket createPacket(Exchange exchange) {
    PlatformPacket packet = new PlatformPacket();

    packet.setExchangeProperties(serializableCopy(exchange.getProperties()));
    packet.setMessageProperties(serializableCopy(exchange.getIn().getHeaders()));

    // create bytes from input body
    // (using data formats or converters)
    byte[] bytes = getInBody(exchange);

    // set body bytes on packet
    packet.setMessageBody(bytes);

    if (outConversion) {
        // set body content from bytes 
        // (using data formats or converters)
        setInBody(bytes, exchange);
    }
    return packet;
}

@SuppressWarnings("unchecked")
@Override
public void process(Exchange exchange) throws Exception {
    Message in = exchange.getIn();
    Address address = in.getHeader(Constants.ADDRESS_HEADER, Address.class);
    Class<?> datatype = in.getHeader(Constants.DATATYPE_HEADER, Class.class);
    Object value = in.getBody(Object.class);
    PlcWriteRequest plcSimpleWriteRequest = new PlcWriteRequest(datatype, address, value);
    PlcWriter plcWriter = plcConnection.getWriter().orElseThrow(() -> new IllegalArgumentException("Writer for driver not found"));
    CompletableFuture<PlcWriteResponse> completableFuture = plcWriter.write(plcSimpleWriteRequest);
    int currentlyOpenRequests = openRequests.incrementAndGet();
    try {
        log.debug("Currently open requests including {}:{}", exchange, currentlyOpenRequests);
        PlcWriteResponse plcWriteResponse = completableFuture.get();
        if (exchange.getPattern().isOutCapable()) {
            Message out = exchange.getOut();
            out.copyFrom(exchange.getIn());
            out.setBody(plcWriteResponse);
        } else {
            in.setBody(plcWriteResponse);
        }
    } finally {
        int openRequestsAfterFinish = openRequests.decrementAndGet();
        log.trace("Open Requests after {}:{}", exchange, openRequestsAfterFinish);
    }
}

@Test
public void singleCity() throws Exception {
    mockGlobalWeatherResponseBody = "<NewDataSet><Table><Country>TEST</Country><City>AA</City></Table></NewDataSet>";

    Exchange response =  fluentProducerTemplate.send();
    String responseBody = response.getIn().getBody(String.class);
    CitiesResponse citiesResponse = objectMapper.readValue(responseBody,CitiesResponse.class);

    assertEquals(200, response.getIn().getHeader(Exchange.HTTP_RESPONSE_CODE));
    assertEquals("TEST", citiesResponse.getCountry());

    List<City> expectedCities = Arrays.asList(
            new City("AA", Arrays.asList("ZIP-AA"))
    );
    log.info("Expected cities: {}",objectMapper.writeValueAsString(expectedCities));
    log.info("Response cities: {}",objectMapper.writeValueAsString(citiesResponse.getCities()));

    assertEquals(expectedCities, citiesResponse.getCities());
}

@Override
public void onCompletion(Exchange exchange) {
    if (wrappedAggregationStrategy != null
            && wrappedAggregationStrategy instanceof CompletionAwareAggregationStrategy) {
        ((CompletionAwareAggregationStrategy) wrappedAggregationStrategy).onCompletion(exchange);
    }

    // Remove exception, fault and redelivery info from exchange
    exchange.setException(null);
    exchange.removeProperty(Exchange.FAILURE_HANDLED);
    exchange.removeProperty(Exchange.FAILURE_ENDPOINT);
    exchange.removeProperty(Exchange.FAILURE_ROUTE_ID);
    exchange.removeProperty(Exchange.ERRORHANDLER_CIRCUIT_DETECTED);
    exchange.removeProperty(Exchange.ERRORHANDLER_HANDLED);
    exchange.removeProperty(Exchange.EXCEPTION_HANDLED);
    exchange.removeProperty(Exchange.EXCEPTION_CAUGHT);

    Message message = exchange.hasOut() ? exchange.getOut() : exchange.getIn();
    message.setFault(false);
    message.removeHeader(Exchange.REDELIVERED);
    message.removeHeader(Exchange.REDELIVERY_COUNTER);
    message.removeHeader(Exchange.REDELIVERY_DELAY);
    message.removeHeader(Exchange.REDELIVERY_EXHAUSTED);
    message.removeHeader(Exchange.REDELIVERY_MAX_COUNTER);
}

@Test
public void shortBusinessId() throws Exception {
    // Call get
    Exchange response = fluentProducerTemplate.to("undertow:http://localhost:{{local.server.port}}/api/user/1")
            .withHeader(Exchange.HTTP_METHOD, HttpMethod.GET)
            .withHeader(Exchange.ACCEPT_CONTENT_TYPE, MediaType.APPLICATION_JSON)
            .withHeader(RestEndpoints.HEADER_BUSINESSID,"X")
            .send();

    assertEquals(500, response.getIn().getHeader(Exchange.HTTP_RESPONSE_CODE));
    assertEquals("X",response.getIn().getHeader(RestEndpoints.HEADER_BUSINESSID,String.class));

    String responseBody = response.getIn().getBody(String.class); //Body is byte[] (or may be InputStream in case of other http clients)
    log.info("body: {}",responseBody);
    ApiResponse apiResponse = objectmapper.readValue(responseBody, ApiResponse.class); //Unmarshall manually with Jackson
    assertEquals("'businessId' length must be between 16 and 48", apiResponse.getMessage());

}

@Override
public void process(Exchange exchange) throws Exception {
  try {
    String requestId = exchange.getProperty(RequestIdManager.REQUEST_ID, String.class);
    ThreadContext.put(RequestIdManager.REQUEST_ID, requestId);
    doProcess(exchange);
  } finally {
    ThreadContext.remove(RequestIdManager.REQUEST_ID);
  }
}

/**
 * Returns true if the body class is a java type which may be converted to a String so we can test for
 * it being JSON already
 */
private boolean isPossibleJsonClass(Exchange exchange, Class<?> clazz, Object body) {
    if (stringableClasses.contains(clazz)) {
        return true;
    }
    for (Class<?> stringableInterface : stringableInterfaces) {
        if (stringableInterface.isInstance(body)) {
            return true;
        }
    }
    return false;
}

public void error(Exchange exchange){
  Object obj = exchange.getProperty(EXCHANGE_PERF_PROPERTY);
  long cost = -1;
  if(obj != null){
    cost = System.currentTimeMillis() - (long) obj;
    exchange.removeProperty(EXCHANGE_PERF_PROPERTY);
  }
  String ex = "";
  if(exchange.getException() != null){
    ex = exchange.getException().getMessage();
    exchange.removeProperty(EXCHANGE_PERF_PROPERTY);
  }
  logger.debug("error{}|{}|{}|{}",ex,cost,exchange.getIn().getHeaders(),exchange.getIn().getBody());
  requestIdManager.clean(exchange);
}

/**
 * Lets marshal the body to JSON using Jackson if we require it.
 * <br>
 * The current rules are to only marshal to JSON if we don't have a {@link Exchange#CONTENT_TYPE} header.
 * If we can convert the body to a String then we test if its already JSON and if not we marshal it using the JSON
 * data format with the Jackson library
 */
public void jsonMarshalIfRequired(Exchange exchange) throws Exception {
    Message in = exchange.getIn();
    if (in == null) {
        return;
    }
    String contentType = in.getHeader(Exchange.CONTENT_TYPE, String.class);
    if (!Strings.isEmpty(contentType)) {
        // lets preserve existing content types as it could be XML, YAML or something else
        return;
    }
    Object body = in.getBody();
    if (body != null) {
        if (isPossibleJsonClass(exchange, body.getClass(), body)) {
            try {
                String text = in.getBody(String.class);
                if (text != null) {
                    if (isJsonLookingString(text.trim())) {
                        in.setHeader(Exchange.CONTENT_TYPE, JSON_CONTENT_TYPE);
                    }
                    in.setBody(text);
                    return;
                }
            } catch (Exception e) {
                // ignore
            }
        }

        in.setHeader(Exchange.CONTENT_TYPE, JSON_CONTENT_TYPE);

        jsonMarshalProducer.process(exchange);
    }
}

public void process(Exchange exchange) throws Exception {
    String key = (String)exchange.getIn().getHeader(WebsocketConstants.CONNECTION_KEY);

    MDC.clear();
    MDC.put("WebsocketConstants.CONNECTION_KEY",key);

    logger.info("Headers: {}",exchange.getIn().getHeaders());
}

private static String mapName(Parameter p){
    String name;
    if(p.getClass().isAssignableFrom(Exchange.class)){
        name = "exchange";
    }else {
        name =  p.getName();
    }
    return name;
}

public void start(Object exchange) {
    if (exchange != null && exchange instanceof Exchange) {
        Exchange exchange1 = (Exchange) exchange;
        Operation uriName = (Operation)exchange1.getIn().getHeader(BeanOperationName);
        if (uriName != null) {
            exchange1.getIn().setHeader(MetricsOperationTimer, createTimerContext(uriName.toString()));
        }
    }

}

@Override
public void onExchangeDone(Route route, Exchange exchange) {
    super.onExchangeDone(route, exchange);

    LOG.info("Exchange Done for route " + route.getId() +
            " exchange: " + exchange.getExchangeId() + " in: " + exchange.getIn().getBody(String.class));
    try {
        stopRoute(route);
    } catch (Exception e) {
        throw new RuntimeExchangeException(e.getMessage(), exchange, e);
    }
}

@Test
public void testSDPParseAnswer() throws Exception {
    Exchange ex = new DefaultExchange(new DefaultCamelContext());
    ex.getIn().setBody(answerSdp);
    processor.process(ex);
    assertEquals(ex.getIn().getBody().getClass(),Answer.class);
}

private Map jsonStringAsMap(String body, Exchange exchange) {
    ObjectMapper mapper = resolveObjectMapper(exchange.getContext().getRegistry());

    // convert JSON string to Map
    try {
        return mapper.readValue(body, new TypeReference<Map<String, Object>>(){});
    } catch (IOException e) {
        // ignore because we are attempting to convert, but its not a JSON document
    }
    return null;
}

@Override
public void onExchangeBegin(Route route, Exchange exchange) {
    super.onExchangeBegin(route, exchange);

    LOG.info("Exchange Begin for route " + route.getId() +
            " exchange: " + exchange.getExchangeId());
}

@Test
public void testSendXMLWhichShouldPassThroughUnchanged() throws Exception {
    String xmlContentType = "text/xml";
    String expectedBody = "<foo>bar</foo>";

    resultEndpoint.expectedBodiesReceived(expectedBody);
    resultEndpoint.allMessages().header(Exchange.CONTENT_TYPE).isEqualTo(xmlContentType);

    template.sendBodyAndHeader(expectedBody, Exchange.CONTENT_TYPE, xmlContentType);

    resultEndpoint.assertIsSatisfied();
}

/**
 * Lets marshal the body to JSON using Jackson if we require it.
 * <br>
 * The current rules are to only marshal to JSON if we don't have a {@link Exchange#CONTENT_TYPE} header.
 * If we can convert the body to a String then we test if its already JSON and if not we marshal it using the JSON
 * data format with the Jackson library
 */
public void jsonMarshalIfRequired(Exchange exchange) throws Exception {
    Message in = exchange.getIn();
    if (in == null) {
        return;
    }
    String contentType = in.getHeader(Exchange.CONTENT_TYPE, String.class);
    if (!Strings.isEmpty(contentType)) {
        // lets preserve existing content types as it could be XML, YAML or something else
        return;
    }
    Object body = in.getBody();
    if (body != null) {
        if (isPossibleJsonClass(exchange, body.getClass(), body)) {
            try {
                String text = in.getBody(String.class);
                if (text != null) {
                    if (isJsonLookingString(text.trim())) {
                        in.setHeader(Exchange.CONTENT_TYPE, JSON_CONTENT_TYPE);
                    }
                    in.setBody(text);
                    return;
                }
            } catch (Exception e) {
                // ignore
            }
        }

        in.setHeader(Exchange.CONTENT_TYPE, JSON_CONTENT_TYPE);

        jsonMarshalProducer.process(exchange);
    }
}

/**
 * Returns true if the body class is a java type which may be converted to a String so we can test for
 * it being JSON already
 */
private boolean isPossibleJsonClass(Exchange exchange, Class<?> clazz, Object body) {
    if (stringableClasses.contains(clazz)) {
        return true;
    }
    for (Class<?> stringableInterface : stringableInterfaces) {
        if (stringableInterface.isInstance(body)) {
            return true;
        }
    }
    return false;
}

protected void logMessagesReceived(MockEndpoint... mockEndpoints) {
    for (MockEndpoint mockEndpoint : mockEndpoints) {
        LOG.info("Messages received on endpoint " + mockEndpoint.getEndpointUri());
        List<Exchange> exchanges = mockEndpoint.getExchanges();
        Assertions.assertThat(exchanges).describedAs("exchanges on " + mockEndpoint).isNotNull();
        int count = 0;
        for (Exchange exchange : exchanges) {
            LOG.info("  " + count++ + " = " + exchange.getIn().getBody(String.class));
        }
    }
}

@Test
public void testSendXMLWhichShouldPassThroughUnchanged() throws Exception {
    String xmlContentType = "text/xml";
    String expectedBody = "<foo>bar</foo>";

    resultEndpoint.expectedBodiesReceived(expectedBody);
    resultEndpoint.allMessages().header(Exchange.CONTENT_TYPE).isEqualTo(xmlContentType);

    template.sendBodyAndHeader(expectedBody, Exchange.CONTENT_TYPE, xmlContentType);

    resultEndpoint.assertIsSatisfied();
}

protected void assertSendBodyReceivesCorrectPayload(Object inputBody, String expectedBody) throws InterruptedException {
    resultEndpoint.expectedBodiesReceived(expectedBody);
    resultEndpoint.allMessages().header(Exchange.CONTENT_TYPE).isEqualTo(JsonEndpoint.JSON_CONTENT_TYPE);

    template.sendBody(inputBody);

    resultEndpoint.assertIsSatisfied();
}

@Test
public void shouldConvertUnifiedToHeadersAndBody() throws Exception {
    final Exchange exchange = createExhangeWithBody(
        "{\"parameters\":{\"slug\":\"1\", \"tick\":\"tock\"},\"body\":{\"description\":\"hello\"}}");

    converter.process(exchange);

    assertThat(exchange.getIn().getHeader("slug")).isEqualTo("1");
    assertThat(exchange.getIn().getHeader("tick")).isEqualTo("tock");
    assertThat(exchange.getIn().getBody()).isEqualTo("{\"description\":\"hello\"}");
}

@SuppressWarnings("unchecked")
@Override
public void process(Exchange exchange) throws Exception {

     LOGGER.debug("Body in (Map): {}", exchange.getIn().getBody());
     String jsonBean = JSONBeanUtil.toJSONBean(exchange.getIn().getBody(Map.class));
     LOGGER.debug("Body out (JSON): {}", jsonBean);
     exchange.getIn().setBody(jsonBean);
}

@Test
public void testConvertCamelPipeline() throws InterruptedException {
    mock1.expectedBodiesReceived("test");
    mock2.expectedBodiesReceived("test");
    Exchange result = producerTemplate.send("direct:flow-test-pipe",
            createMessage("test").getExchange());
    mock1.assertIsSatisfied();
    mock2.assertIsSatisfied();
    assertEquals(initial, new PlatformMessage(mock1.getExchanges().get(0)).getSplitHistory());
    assertEquals(initial, new PlatformMessage(mock2.getExchanges().get(0)).getSplitHistory());
    assertEquals(initial, new PlatformMessage(result).getSplitHistory()); 
}

private Exchange processResult(final Exchange origExchange,
                               final SplitIndex index,
                               final Object splitPart) throws Exception {

    final Exchange subExchange = origExchange.copy();

    Message message = subExchange.getIn();
    message.setBody(splitPart);
    finalizeSubExchange(origExchange, subExchange, index);

    super.processNext(subExchange);
    return subExchange;
}

public ActiveMQPublishTextComponent() {
    super("activemq-publish-text", ActiveMQPublishTextComponent.class.getName());

    setBeforeProducer( (Exchange exchange) -> {

        // extract headers and body
        Message out = exchange.getIn();
        JmsTextMessage jmsTextMessage = out.getBody(JmsTextMessage.class);
        out.setBody(jmsTextMessage.getBody());
        if (jmsTextMessage.getHeaders() != null) {
            out.setHeaders(jmsTextMessage.getHeaders());
        }
    });
}