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

@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 {
    Message in = exchange.getIn();

    Status status = exchange.getIn().getBody(Status.class);

    User user = status.getUser();
    String name = user.getName();
    String screenName = user.getScreenName();

    Contact contact = new Contact();
    contact.setLastName(name);
    contact.setTwitterScreenName__c(screenName);

    in.setBody(contact);
}

@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);
    }
}

@Override
public void process(Exchange exchange) throws Exception {
    Message in = exchange.getIn();

    String actionName = in.getHeader(SpongeConstants.SPONGE_ACTION, String.class);
    if (actionName != null) {
        // Remove the header so it won't be propagated.
        in.removeHeader(SpongeConstants.SPONGE_ACTION);
    }

    if (actionName == null) {
        actionName = action != null ? action : CamelProducerAction.NAME;
    }

    Object result = engine.getOperations().call(actionName, exchange);

    exchange.getIn().setBody(result);
}

@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);
}

@Override
public void process(Exchange exchange) throws Exception {
    OrientDBEndpoint endpoint = (OrientDBEndpoint)getEndpoint();
    curDb = endpoint.databaseOpen();
    Object input = exchange.getIn().getBody();
    Message out = exchange.getOut(); 
    out.getHeaders().putAll(exchange.getIn().getHeaders());


    if (input instanceof List){
        out.setBody(endpoint.makeOutObject(processList((List<?>)input, endpoint, curDb)));
    }else if (input instanceof String && isJSONList((String)input)){
        List<String> inputList =  strToJSONsList((String)input);
        out.setBody(endpoint.makeOutObject(processList(inputList, endpoint, curDb)));
    }else{
        out.setBody(endpoint.makeOutObject(processSingleObject(input, endpoint, curDb)));
    }
    endpoint.databaseClose(curDb);
    curDb=null;
}

private void bind(InvokeOnHeader handler, final Method method) {
    if (handler != null && method.getParameterCount() == 1) {
        method.setAccessible(true);

        final Class<?> type = method.getParameterTypes()[0];

        LOGGER.debug("bind key={}, class={}, method={}, type={}",
            handler.value(), this.getClass(), method.getName(), type);

        if (Message.class.isAssignableFrom(type)) {
            bind(handler.value(), e -> method.invoke(target, e.getIn()));
        } else {
            bind(handler.value(), e -> method.invoke(target, e));
        }
    }
}

@Test
public void testTheCopyOfHeaders() {
  Message msg = new DefaultMessage();
  msg.setHeader("CamelRedelivered", false);
  msg.setHeader("CamelRedeliveryCounter", 0);
  msg.setHeader("JMSCorrelationID", "");
  msg.setHeader("JMSDestination", "queue://dev.msy.queue.log.fwd");
  msg.setHeader("JMSReplyTo", null);

  DeliveryOptions options = CamelHelper.getDeliveryOptions(msg, true);

  assertThat(options.getHeaders().get("CamelRedelivered")).isEqualToIgnoringCase("false");
  assertThat(options.getHeaders().get("CamelRedeliveryCounter")).isEqualToIgnoringCase("0");
  assertThat(options.getHeaders().get("JMSCorrelationID")).isEqualToIgnoringCase("");
  assertThat(options.getHeaders().get("JMSDestination")).isEqualToIgnoringCase("queue://dev.msy.queue.log.fwd");
  assertThat(options.getHeaders().get("JMSReplyTo")).isNull();

}

public <T> T getOut(Class<T> type) {
    if (!hasOut()) {
        return null;
    }

    Message out = getOut();

    // eager same instance type test to avoid the overhead of invoking the type converter
    // if already same type
    if (type.isInstance(out)) {
        return type.cast(out);
    }

    // fallback to use type converter
    return context.getTypeConverter().convertTo(type, this, out);
}

private Element getParentForDetachedCase(Document doc, Message inMessage, String referenceUri) throws XmlSignatureException {
    String elementId = referenceUri;
    if (elementId.startsWith("#")) {
        elementId = elementId.substring(1);
    }
    Element el = doc.getElementById(elementId);
    if (el == null) {
        // should not happen because has been checked before
        throw new IllegalStateException("No element found for element ID " + elementId);
    }
    LOG.debug("Sibling element of the detached XML Signature with reference URI {}: {}  {} ",
            new Object[] {referenceUri, el.getLocalName(), el.getNamespaceURI() });
    Element result = getParentElement(el);
    if (result != null) {
        return result;
    } else {
        throw new XmlSignatureException(
                "Either the configuration of the XML Signature component is wrong or the incoming document has an invalid structure: The element "
                        + el.getLocalName() + "{" + el.getNamespaceURI() + "} which is referenced by the reference URI " + referenceUri
                        + " has no parent element. The element must have a parent element in the configured detached case.");
    }
}

protected RouteBuilder createRouteBuilder() {
    return new RouteBuilder() {
        public void configure() {
            errorHandler(noErrorHandler());
            from(getFromEndpointUri()).process(new Processor() {
                public void process(final Exchange exchange) {
                    Message in = exchange.getIn();
                    Node node = in.getBody(Node.class);
                    assertNotNull(node);
                    XmlConverter xmlConverter = new XmlConverter();
                    // Put the result back
                    exchange.getOut().setBody(xmlConverter.toSource(RESPONSE));
                }
            });
        }
    };
}

protected RouteBuilder createRouteBuilder() throws Exception {
    return new RouteBuilder() {
        public void configure() throws Exception {

            // Test the filter list options
            from("jetty://http://localhost:{{port}}/testFilters?filtersRef=myFilters&filterInit.keyWord=KEY").process(new Processor() {
                public void process(Exchange exchange) throws Exception {
                    Message in = exchange.getIn();
                    String request = in.getBody(String.class);
                    // The other form date can be get from the message header
                    exchange.getOut().setBody(request + " response");
                }
            });
        }
    };
}

public void process(Exchange exchange) throws Exception {
    Message inMessage = exchange.getIn();                        
    // Get the operation name from in message
    String operationName = inMessage.getHeader(CxfConstants.OPERATION_NAME, String.class);
    if ("getCustomer".equals(operationName)) {
        processGetCustomer(exchange);
    } else if ("updateCustomer".equals(operationName)) {
        assertEquals("Get a wrong customer message header", "header1;header2", inMessage.getHeader("test"));
        String httpMethod = inMessage.getHeader(Exchange.HTTP_METHOD, String.class);
        assertEquals("Get a wrong http method", "PUT", httpMethod);
        Customer customer = inMessage.getBody(Customer.class);
        assertNotNull("The customer should not be null.", customer);
        // Now you can do what you want on the customer object
        assertEquals("Get a wrong customer name.", "Mary", customer.getName());
        // set the response back
        exchange.getOut().setBody(Response.ok().build());
    }

}

protected void updateMessageHeader(Message in, ChannelHandlerContext ctx) {
    in.setHeader(NettyConstants.NETTY_CHANNEL_HANDLER_CONTEXT, ctx);
    in.setHeader(NettyConstants.NETTY_REMOTE_ADDRESS, ctx.channel().remoteAddress());
    in.setHeader(NettyConstants.NETTY_LOCAL_ADDRESS, ctx.channel().localAddress());

    if (configuration.isSsl()) {
        // setup the SslSession header
        SSLSession sslSession = getSSLSession(ctx);
        in.setHeader(NettyConstants.NETTY_SSL_SESSION, sslSession);

        // enrich headers with details from the client certificate if option is enabled
        if (configuration.isSslClientCertHeaders()) {
            enrichWithClientCertInformation(sslSession, in);
        }
    }
}

private String parentFolderPathFor(Message message) throws Exception {
    String customPath = message.getHeader(CamelCMISConstants.CMIS_FOLDER_PATH, String.class);
    if (customPath != null) {
        return customPath;
    }

    if (isFolder(message)) {
        String path = (String) message.getHeader(PropertyIds.PATH);
        String name = (String) message.getHeader(PropertyIds.NAME);
        if (path != null && path.length() > name.length()) {
            return path.substring(0, path.length() - name.length());
        }
    }

    return "/";
}

protected Element getParentOfSignature(Message inMessage, Node messageBodyNode, String contentReferenceURI, SignatureType sigType)
    throws Exception { //NOPMD
    if (SignatureType.enveloping == sigType) {
        // enveloping case
        return null;
    }
    if (messageBodyNode.getParentNode() == null || messageBodyNode.getParentNode().getNodeType() != Node.DOCUMENT_NODE) {
        throw new XmlSignatureFormatException(
                "Incomming message has wrong format: It is not an XML document. Cannot create an enveloped or detached XML signature.");
    }
    Document doc = (Document) messageBodyNode.getParentNode();
    if (SignatureType.detached == sigType) {
        return getParentForDetachedCase(doc, inMessage, contentReferenceURI);
    } else {
        // enveloped case
        return getParentForEnvelopedCase(doc, inMessage);
    }

}

protected void doWriteGZIPResponse(Message message, HttpServletResponse response, Exchange exchange) throws IOException {
    byte[] bytes;
    try {
        bytes = message.getMandatoryBody(byte[].class);
    } catch (InvalidPayloadException e) {
        throw ObjectHelper.wrapRuntimeCamelException(e);
    }

    byte[] data = GZIPHelper.compressGZIP(bytes);
    ServletOutputStream os = response.getOutputStream();
    try {
        if (LOG.isDebugEnabled()) {
            LOG.debug("Streaming response as GZIP in non-chunked mode with content-length {} and buffer size: {}", data.length, response.getBufferSize());
        }
        response.setContentLength(data.length);
        os.write(data);
        os.flush();
    } finally {
        IOHelper.close(os);
    }
}

/**
 * Copies the headers from the source to the target message.
 * 
 * @param source the source message
 * @param target the target message
 * @param strategy the header filter strategy which could help us to filter the protocol message headers
 * @param override whether to override existing headers
 */
public static void copyHeaders(Message source, Message target, HeaderFilterStrategy strategy, boolean override) {
    if (!source.hasHeaders()) {
        return;
    }

    for (Map.Entry<String, Object> entry : source.getHeaders().entrySet()) {
        String key = entry.getKey();
        Object value = entry.getValue();

        if (target.getHeader(key) == null || override) {
            if (strategy == null) {
                target.setHeader(key, value);
            } else if (!strategy.applyFilterToExternalHeaders(key, value, target.getExchange())) {
                // Just make sure we don't copy the protocol headers to target
                target.setHeader(key, value);
            }
        }
    }
}

@Test
public void testGrabbingListOfEntries() throws Exception {
    MockEndpoint mock = getMockEndpoint("mock:result");
    // as the default delay option is one hour long, we expect exactly one message exchange
    mock.expectedMessageCount(1);

    template.sendBody("direct:start", "Hello World");

    mock.assertIsSatisfied();

    Exchange exchange = mock.getExchanges().get(0);
    assertNotNull(exchange);
    Message in = exchange.getIn();
    assertNotNull(in);
    String weather = assertIsInstanceOf(String.class, in.getBody());

    checkWeatherContent(weather);
}

@Test
public void testConvertPayloadToInputStream() throws Exception {
    MockEndpoint mockEndpoint = getMockEndpoint("mock:result");
    mockEndpoint.expectedMessageCount(1);

    template.requestBodyAndHeader("netty4-http:http://localhost:{{port}}/test", expectedBody, "Content-Type", "application/xml");

    mockEndpoint.assertIsSatisfied();
    List<Exchange> list = mockEndpoint.getReceivedExchanges();
    Exchange exchange = list.get(0);
    assertNotNull("exchange", exchange);

    Message in = exchange.getIn();
    assertNotNull("in", in);

    Object actual = in.getBody();
    InputStream value = assertIsInstanceOf(InputStream.class, actual);
    assertNotNull("InputStream", value);
}

/**
 * Gets the Camel {@link Message} to use as the message to be set on the current {@link Exchange} when
 * we have received a reply message.
 * <p/>
 *
 * @param exchange      the current exchange
 * @param messageEvent  the incoming event which has the response message from Netty.
 * @return the Camel {@link Message} to set on the current {@link Exchange} as the response message.
 * @throws Exception is thrown if error getting the response message
 */
protected Message getResponseMessage(Exchange exchange, MessageEvent messageEvent) throws Exception {
    Object body = messageEvent.getMessage();
    if (LOG.isDebugEnabled()) {
        LOG.debug("Channel: {} received body: {}", new Object[]{messageEvent.getChannel(), body});
    }

    // if textline enabled then covert to a String which must be used for textline
    if (producer.getConfiguration().isTextline()) {
        body = producer.getContext().getTypeConverter().mandatoryConvertTo(String.class, exchange, body);
    }

    // set the result on either IN or OUT on the original exchange depending on its pattern
    if (ExchangeHelper.isOutCapable(exchange)) {
        NettyPayloadHelper.setOut(exchange, body);
        return exchange.getOut();
    } else {
        NettyPayloadHelper.setIn(exchange, body);
        return exchange.getIn();
    }
}

@Override
public boolean process(Exchange exchange, AsyncCallback callback) {
    try {
        Object newHeader = expression.evaluate(exchange, Object.class);

        if (exchange.getException() != null) {
            // the expression threw an exception so we should break-out
            callback.done(true);
            return true;
        }

        boolean out = exchange.hasOut();
        Message old = out ? exchange.getOut() : exchange.getIn();

        String key = headerName.evaluate(exchange, String.class);
        old.setHeader(key, newHeader);

    } catch (Throwable e) {
        exchange.setException(e);
    }

    callback.done(true);
    return true;
}

/**
 * Transfers path parameters from the full path (including ancestor subresource locators) into Camel IN Message Headers.
 */
@SuppressWarnings("unchecked")
protected void bindHeadersFromSubresourceLocators(Exchange cxfExchange, org.apache.camel.Exchange camelExchange) {
    MultivaluedMap<String, String> pathParams = (MultivaluedMap<String, String>) 
            cxfExchange.getInMessage().get(URITemplate.TEMPLATE_PARAMETERS);

    // return immediately if we have no path parameters
    if (pathParams == null || (pathParams.size() == 1 && pathParams.containsKey(URITemplate.FINAL_MATCH_GROUP))) {
        return;
    }

    Message m = camelExchange.getIn();
    for (Entry<String, List<String>> entry : pathParams.entrySet()) {
        // skip over the FINAL_MATCH_GROUP which stores the entire path
        if (URITemplate.FINAL_MATCH_GROUP.equals(entry.getKey())) {
            continue;
        }
        m.setHeader(entry.getKey(), entry.getValue().get(0));
    }
}

private void assertExchange(Exchange exchange, boolean hasFault) {
    if (!hasFault) {
        Message out = exchange.getOut();
        assertNotNull(out);
        assertFalse(out.isFault());
        assertEquals("Goodbye!", out.getBody());
        assertEquals("cheddar", out.getHeader("cheese"));
    } else {
        Message fault = exchange.getOut();
        assertNotNull(fault);
        assertTrue(fault.isFault());
        assertNotNull(fault.getBody());
        assertTrue("Should get the InterrupteException exception", fault.getBody() instanceof InterruptedException);
        assertEquals("nihao", fault.getHeader("hello"));
    }


    // in should stay the same
    Message in = exchange.getIn();
    assertNotNull(in);
    assertEquals("Hello!", in.getBody());
    assertEquals("feta", in.getHeader("cheese"));
    // however the shared properties have changed
    assertEquals("fresh", exchange.getProperty("salami"));
    assertNull(exchange.getProperty("Charset"));
}

/**
 * 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);
    }
}

@Override
public void process(final Exchange exchange) throws Exception {
    final Message in = exchange.getIn();

    final String body = in.getBody(String.class);
    if (body == null) {
        return;
    }

    final JsonNode payload;
    try {
        payload = MAPPER.readTree(body);
    } catch (final JsonProcessingException e) {
        LOG.warn("Unable to parse payload, continuing without conversion", e);

        return;
    }

    payload.with("parameters").fields().forEachRemaining(e -> in.setHeader(e.getKey(), e.getValue().asText()));

    final JsonNode requestBody = payload.get("body");
    if (requestBody == null) {
        in.setBody(null);
    } else {
        in.setBody(MAPPER.writeValueAsString(requestBody));
    }
}

public AbstractODataEntityConnector(String componentName, String componentScheme, String className) {
    super(componentName, componentScheme, className);

    setBeforeProducer(exchange -> {
        // convert json into ClientEntity
        Message in = exchange.getIn();
        ignoreResponseHeaders(in);
        final ODataBinder binder = odataClient.getBinder();
        final ClientODataDeserializer deserializer = odataClient.getDeserializer(ContentType.APPLICATION_JSON);
        ClientEntity oDataEntity = binder.getODataEntity(deserializer.toEntity(in.getBody(InputStream.class)));
        in.setBody(oDataEntity);
    });
}

public AbstractODataResourceConnector(String componentName, String componentScheme, String className) {
    super(componentName, componentScheme, className);

    // replace DTO with headers
    setBeforeProducer(exchange -> {
        final Message in = exchange.getIn();
        ignoreResponseHeaders(in);
        final ODataResource resource = in.getBody(ODataResource.class);
        in.setHeader(Olingo4Constants.PROPERTY_PREFIX + "keyPredicate", resource.getKeyPredicate());
        in.setBody(null);
    });
}

protected SalesforceConnectorExpectingId(final String componentName, final String componentScheme, final Class<?> componentClass) {
    super(componentName, componentScheme, componentClass);

    // replace DTO with id for Salesforce component
    setBeforeProducer(exchange -> {
        final Message in = exchange.getIn();
        final SalesforceIdentifier id = in.getBody(SalesforceIdentifier.class);

        if (id != null) {
            in.setBody(id.getId());
        }
    });
}

@Override
public void process(final Exchange exchange) throws Exception {
    // parse input json and extract Id field
    final Message in = exchange.getIn();
    final String body = in.getBody(String.class);

    if (body == null) {
        return;
    }

    final ObjectNode node = (ObjectNode) MAPPER.readTree(body);

    final String idPropertyName = determineIdProperty(exchange);

    final JsonNode idProperty = node.remove(idPropertyName);
    if (idProperty == null) {
        exchange.setException(
            new SalesforceException("Missing option value for Id or " + SalesforceEndpointConfig.SOBJECT_EXT_ID_NAME, 404));

        return;
    }

    final String idValue = idProperty.textValue();
    if ("Id".equals(idPropertyName)) {
        in.setHeader(SalesforceEndpointConfig.SOBJECT_ID, idValue);
    } else {
        in.setHeader(SalesforceEndpointConfig.SOBJECT_EXT_ID_VALUE, idValue);
    }

    // base fields are not allowed to be updated
    clearBaseFields(node);

    // update input json
    in.setBody(MAPPER.writeValueAsString(node));
}

@Test
public void shouldAllowNullInput() throws Exception {
    final Message in = exchange.getIn();
    beforeProducer.process(exchange);

    assertThat(in.getBody()).isNull();
}

@Test
public void shouldAllowNullOutput() throws Exception {
    final Message out = exchange.getOut();
    afterProducer.process(exchange);

    assertThat(out.getBody()).isNull();
}

@Test
public void shouldNotConvertFailedExchanges() throws Exception {
    final Message out = exchange.getOut();
    out.setBody("wat");

    exchange.setException(new Exception());
    afterProducer.process(exchange);

    assertThat(out.getBody()).isEqualTo("wat");
}

@Test
public void shouldUnmarshallToSpecifiedInputType() throws Exception {
    final Message in = exchange.getIn();
    in.setBody("{}");

    beforeProducer.process(exchange);

    assertThat(in.getBody()).isInstanceOf(SalesforceIdentifier.class);
}

@Test
public void shouldUnmarshallToSpecifiedOutputType() throws Exception {
    final Message out = exchange.getOut();
    out.setBody("{}");

    afterProducer.process(exchange);

    assertThat(out.getBody()).isInstanceOf(AbstractDTOBase.class);
}

public void process(Exchange exchange) throws Exception {

    Message in = exchange.getIn();
    Map<String,Object> tuple= (Map<String, Object>) in.getBody();
    MoneyTransfer mt = new  MoneyTransfer();
    mt.setId((Long)tuple.get(MoneyTransfer.FIELD_ID));
    mt.setKeycode((String)tuple.get(MoneyTransfer.FIELD_KEYCODE));
    mt.setPaymentMethod((String)tuple.get(MoneyTransfer.FIELD_PAYMENT_METHOD));
    mt.setAmount_hf_sender_currency((BigDecimal)tuple.get(MoneyTransfer.FIELD_AMOUNT_HF_SENDER_CUR));
    mt.setStatus((String)tuple.get(MoneyTransfer.FIELD_STATUS));
    in.setBody(mt);
    in.setHeader(InfinispanConstants.KEY, mt.getKeycode());

}

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());
        }
    });
}

public ActiveMQSubscribeMapComponent() {
    super("activemq-subscribe-map", ActiveMQSubscribeMapComponent.class.getName());

    // create JmsMessage from Camel message
    setBeforeConsumer(exchange -> {
        final Message in = exchange.getIn();
        final JmsMapMessage jmsMapMessage = new JmsMapMessage(in.getBody(Map.class));
        jmsMapMessage.setHeaders(in.getHeaders());
        in.setBody(jmsMapMessage);
    });
}

public ActiveMQSubscribeTextComponent() {
    super("activemq-subscribe-text", ActiveMQSubscribeTextComponent.class.getName());

    // create JmsMessage from Camel message
    setBeforeConsumer(exchange -> {
        final Message in = exchange.getIn();
        final JmsTextMessage jmsTextMessage = new JmsTextMessage(in.getBody(String.class));
        jmsTextMessage.setHeaders(in.getHeaders());
        in.setBody(jmsTextMessage);
    });
}

public ActiveMQPublishBytesComponent() {
    super("activemq-publish-bytes", ActiveMQPublishBytesComponent.class.getName());

    setBeforeProducer( (Exchange exchange) -> {

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