Java 类org.apache.camel.util.ExchangeHelper 实例源码

@Override
public boolean matches(Exchange exchange) {
    Object msgBody = exchange.getIn().getBody();

    // TODO: Maybe check for content-type, too ?
    // String contentType = exchange.getIn().getHeader(Exchange.CONTENT_TYPE, String.class);
    // if ("application/json".equals(contentType)) { ... }
    // ???
    if (!(msgBody instanceof String)) {
        return language.createPredicate(expression).matches(exchange);
    }

    Exchange exchangeToCheck = exchange;
    // If it is a json document , suppose that this is a document which needs to be parsed as JSON
    // Therefor we set a map instead of the string
    Map jsonDocument = jsonStringAsMap((String) msgBody, exchange);
    if (jsonDocument != null) {
        // Clone the exchange and set the JSON message converted to a Map / List as in message.
        // The intention is that only this predicate acts on the converted value,
        // but the original in-message still continues to carry the same format
        // The predicated is supposed to be read only with respect to the incoming messaeg.
        exchangeToCheck = ExchangeHelper.createCopy(exchange, true);
        exchangeToCheck.getIn().setBody(jsonDocument);
    }
    return language.createPredicate(convertSimpleToOGNLForMaps(expression)).matches(exchangeToCheck);
}

@Override
public boolean matches(Exchange exchange) {
    Object msgBody = exchange.getIn().getBody();

    // TODO: Maybe check for content-type, too ?
    // String contentType = exchange.getIn().getHeader(Exchange.CONTENT_TYPE, String.class);
    // if ("application/json".equals(contentType)) { ... }
    // ???
    if (!(msgBody instanceof String)) {
        return language.createPredicate(expression).matches(exchange);
    }

    Exchange exchangeToCheck = exchange;
    // If it is a json document , suppose that this is a document which needs to be parsed as JSON
    // Therefor we set a map instead of the string
    Map jsonDocument = jsonStringAsMap((String) msgBody, exchange);
    if (jsonDocument != null) {
        // Clone the exchange and set the JSON message converted to a Map / List as in message.
        // The intention is that only this predicate acts on the converted value,
        // but the original in-message still continues to carry the same format
        // The predicated is supposed to be read only with respect to the incoming messaeg.
        exchangeToCheck = ExchangeHelper.createCopy(exchange, true);
        exchangeToCheck.getIn().setBody(jsonDocument);
    }
    return language.createPredicate(convertSimpleToOGNLForMaps(expression)).matches(exchangeToCheck);
}

public void push(ServerSession remote, ServerMessage cometdMessage) throws Exception {
    Object data = null;

    Message message = binding.createCamelMessage(remote, cometdMessage, data);

    Exchange exchange = endpoint.createExchange();
    exchange.setIn(message);

    consumer.getProcessor().process(exchange);

    if (ExchangeHelper.isOutCapable(exchange)) {
        ServerChannel channel = getBayeux().getChannel(channelName);
        ServerSession serverSession = getServerSession();

        ServerMessage.Mutable outMessage = binding.createCometdMessage(channel, serverSession, exchange.getOut());
        remote.deliver(serverSession, outMessage);
    }
}

protected void maybeDisconnectOnDone(Exchange exchange) {
    if (session == null) {
        return;
    }

    // should session be closed after complete?
    Boolean close;
    if (ExchangeHelper.isOutCapable(exchange)) {
        close = exchange.getOut().getHeader(MinaConstants.MINA_CLOSE_SESSION_WHEN_COMPLETE, Boolean.class);
    } else {
        close = exchange.getIn().getHeader(MinaConstants.MINA_CLOSE_SESSION_WHEN_COMPLETE, Boolean.class);
    }

    // should we disconnect, the header can override the configuration
    boolean disconnect = getEndpoint().getConfiguration().isDisconnect();
    if (close != null) {
        disconnect = close;
    }
    if (disconnect) {
        if (LOG.isDebugEnabled()) {
            LOG.debug("Closing session when complete at address: {}", getEndpoint().getAddress());
        }
        session.close();
    }
}

public synchronized void failedExchange(Exchange exchange) {
    increment();
    exchangesFailed.increment();
    exchangesInflight.decrement();

    if (ExchangeHelper.isRedelivered(exchange)) {
        redeliveries.increment();
    }
    Boolean externalRedelivered = exchange.isExternalRedelivered();
    if (externalRedelivered != null && externalRedelivered) {
        externalRedeliveries.increment();
    }

    long now = new Date().getTime();
    if (firstExchangeFailureTimestamp.getUpdateCount() == 0) {
        firstExchangeFailureTimestamp.updateValue(now);
    }

    lastExchangeFailureTimestamp.updateValue(now);
    if (firstExchangeFailureExchangeId == null) {
        firstExchangeFailureExchangeId = exchange.getExchangeId();
    }
    lastExchangeFailureExchangeId = exchange.getExchangeId();
}

public void process(Exchange exchange) throws Exception {
    String user = exchange.getIn().getHeader("User", String.class);
    String contentType = ExchangeHelper.getContentType(exchange);
    String body = exchange.getIn().getBody(String.class);
    String encoding = exchange.getIn().getHeader(Exchange.CONTENT_ENCODING, String.class);
    if (encoding != null) {
        exchange.getOut().setHeader(Exchange.CONTENT_ENCODING, encoding);
    }
    if ("Claus".equals(user) && contentType.startsWith("text/xml") && body.equals("<order>123</order>")) {
        assertEquals("test", exchange.getIn().getHeader("SOAPAction", String.class));
        if (contentType.endsWith("UTF-8")) {
            assertEquals("Get a wrong charset name.", exchange.getProperty(Exchange.CHARSET_NAME, String.class), "UTF-8");
        }
        exchange.getOut().setBody("<order>OK</order>");
        exchange.getOut().setHeader("Content-Type", "text/xml");
    } else {
        exchange.getOut().setBody("FAIL");
        exchange.getOut().setHeader(Exchange.CONTENT_TYPE, "text/plain");
    }
}

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

@SuppressWarnings("unchecked")
public <T> T getProperty(String name, Class<T> type) {
    Object value = getProperty(name);
    if (value == null) {
        // lets avoid NullPointerException when converting to boolean for null values
        if (boolean.class.isAssignableFrom(type)) {
            return (T) Boolean.FALSE;
        }
        return null;
    }

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

    return ExchangeHelper.convertToType(this, type, value);
}

/**
 * Writes the image file to the output stream.
 *
 * @param graph    the object graph
 * @param exchange the camel exchange
 * @param stream   the output stream
 */
private void printImage(final Exchange exchange, final Object graph, final OutputStream stream) throws Exception {
    final String payload = ExchangeHelper
            .convertToMandatoryType(exchange, String.class, graph);
    final MultiFormatWriter writer = new MultiFormatWriter();

    // set values
    final String type = this.params.getType().toString();

    // create code image  
    final BitMatrix matrix = writer.encode(
            payload,
            this.params.getFormat(),
            this.params.getWidth(),
            this.params.getHeight(),
            writerHintMap);

    // write image back to stream
    MatrixToImageWriter.writeToStream(matrix, type, stream);
}

/**
 * Returns an expression for the outbound message headers
 *
 * @return an expression object which will return the headers, will be <tt>null</tt> if the
 * exchange is not out capable.
 */
public static Expression outHeadersExpression() {
    return new ExpressionAdapter() {
        public Object evaluate(Exchange exchange) {
            // only get out headers if the MEP is out capable
            if (ExchangeHelper.isOutCapable(exchange)) {
                return exchange.getOut().getHeaders();
            } else {
                return null;
            }
        }

        @Override
        public String toString() {
            return "outHeaders";
        }
    };
}

protected void assertMessageExpected(long index, Exchange expected, Exchange actual) throws Exception {
    switch (getDataSetIndex()) {
    case "off":
        break;
    case "strict":
        long actualCounter = ExchangeHelper.getMandatoryHeader(actual, Exchange.DATASET_INDEX, Long.class);
        assertEquals("Header: " + Exchange.DATASET_INDEX, index, actualCounter, actual);
        break;
    case "lenient":
    default:
        // Validate the header value if it is present
        Long dataSetIndexHeaderValue = actual.getIn().getHeader(Exchange.DATASET_INDEX, Long.class);
        if (dataSetIndexHeaderValue != null) {
            assertEquals("Header: " + Exchange.DATASET_INDEX, index, dataSetIndexHeaderValue, actual);
        } else {
            // set the header if it isn't there
            actual.getIn().setHeader(Exchange.DATASET_INDEX, index);
        }
        break;
    }

    getDataSet().assertMessageExpected(this, expected, actual, index);
}

protected synchronized void onExchange(Exchange exchange) {
    try {
        if (reporter != null) {
            reporter.process(exchange);
        }
        Exchange copy = exchange;
        if (copyOnExchange) {
            // copy the exchange so the mock stores the copy and not the actual exchange
            copy = ExchangeHelper.createCopy(exchange, true);
        }
        performAssertions(exchange, copy);
    } catch (Throwable e) {
        // must catch java.lang.Throwable as AssertionError extends java.lang.Error
        failures.add(e);
    } finally {
        // make sure latch is counted down to avoid test hanging forever
        if (latch != null) {
            latch.countDown();
        }
    }
}

private void doProcessResult(Exchange exchange, Object result) {
    if (endpoint.getStatementType() == StatementType.QueryForList || endpoint.getStatementType() == StatementType.QueryForObject) {
        Message answer = exchange.getIn();
        if (ExchangeHelper.isOutCapable(exchange)) {
            answer = exchange.getOut();
            // preserve headers
            answer.getHeaders().putAll(exchange.getIn().getHeaders());
        }
        // set the result as body for insert
        answer.setBody(result);

        answer.setHeader(IBatisConstants.IBATIS_RESULT, result);
        answer.setHeader(IBatisConstants.IBATIS_STATEMENT_NAME, statement);
    } else {
        Message msg = exchange.getIn();
        msg.setHeader(IBatisConstants.IBATIS_RESULT, result);
        msg.setHeader(IBatisConstants.IBATIS_STATEMENT_NAME, statement);
    }
}

protected void maybeDisconnectOnDone(Exchange exchange) {
    if (session == null) {
        return;
    }

    // should session be closed after complete?
    Boolean close;
    if (ExchangeHelper.isOutCapable(exchange)) {
        close = exchange.getOut().getHeader(Mina2Constants.MINA_CLOSE_SESSION_WHEN_COMPLETE, Boolean.class);
    } else {
        close = exchange.getIn().getHeader(Mina2Constants.MINA_CLOSE_SESSION_WHEN_COMPLETE, Boolean.class);
    }

    // should we disconnect, the header can override the configuration
    boolean disconnect = getEndpoint().getConfiguration().isDisconnect();
    if (close != null) {
        disconnect = close;
    }
    if (disconnect) {
        LOG.debug("Closing session when complete at address: {}", address);
        session.close(true);
    }
}

protected void assertCorrectMapReceived() {
    Exchange exchange = endpoint.getReceivedExchanges().get(0);
    // This should be a JMS Exchange
    assertNotNull(ExchangeHelper.getBinding(exchange, JmsBinding.class));
    JmsMessage in = (JmsMessage) exchange.getIn();
    assertNotNull(in);

    Map<?, ?> map = exchange.getIn().getBody(Map.class);
    log.info("Received map: " + map);

    assertNotNull("Should have received a map message!", map);
    assertIsInstanceOf(MapMessage.class, in.getJmsMessage());
    assertEquals("map.foo", "abc", map.get("foo"));
    assertEquals("map.bar", "xyz", map.get("bar"));
    assertEquals("map.size", 2, map.size());
}

@Override
protected RouteBuilder createRouteBuilder() {
    return new RouteBuilder() {
        @Override
        public void configure() throws Exception {
            // enable POST support
            from("restlet:http://localhost:" + portNum + "/users/?restletMethods=post")
                .process(new Processor() {
                    public void process(Exchange exchange) throws Exception {
                        String type = ExchangeHelper.getContentType(exchange);
                        assertEquals("text/xml", type);
                        exchange.getOut().setBody("<status>OK</status>");
                        exchange.getOut().setHeader(Exchange.CONTENT_TYPE, "text/xml");
                    }
                });
        }
    };
}

@Override
protected void onExchange(Exchange exchange) throws Exception {
    StringWriter buffer = new StringWriter();

    @SuppressWarnings("unchecked")
    Map<String, Object> variableMap = exchange.getIn().getHeader(StringTemplateConstants.STRINGTEMPLATE_VARIABLE_MAP, Map.class);
    if (variableMap == null) {
        variableMap = ExchangeHelper.createVariableMap(exchange);
    }

    // getResourceAsInputStream also considers the content cache
    String text = exchange.getContext().getTypeConverter().mandatoryConvertTo(String.class, getResourceAsInputStream());
    ST template = new ST(text, delimiterStart, delimiterStop);
    for (Map.Entry<String, Object> entry : variableMap.entrySet()) {
        template.add(entry.getKey(), entry.getValue());
    }
    log.debug("StringTemplate is writing using attributes: {}", variableMap);
    template.write(new NoIndentWriter(buffer));

    // now lets output the results to the exchange
    Message out = exchange.getOut();
    out.setBody(buffer.toString());
    out.setHeaders(exchange.getIn().getHeaders());
    out.setHeader(StringTemplateConstants.STRINGTEMPLATE_RESOURCE_URI, getResourceUri());
    out.setAttachments(exchange.getIn().getAttachments());
}

public void calculateSignature(Exchange exchange, Signature signer) throws Exception {
    Object payload = exchange.getIn().getBody();
    if (payload != null) {
        InputStream payloadStream = ExchangeHelper.convertToMandatoryType(exchange, InputStream.class, payload);
        try {
            byte[] buffer = new byte[config.getBufferSize()];
            int read;
            while ((read = payloadStream.read(buffer)) > 0) {
                signer.update(buffer, 0, read);
            }
        } finally {
            IOHelper.close(payloadStream);
        }
    }
}

private void processUsersRequest(Exchange exchange) throws Exception {
    final String operation;

    if (endpoint.getOperation() != null) {
        operation = endpoint.getOperation();
    } else {
        operation = ExchangeHelper.getMandatoryHeader(exchange, AlmaMessage.Header.Operation, String.class);
    }

    switch (operation) {
    case AlmaMessage.Operation.Read:
        userService.getUser(exchange);
        break;
    case AlmaMessage.Operation.Create:
        userService.createUser(exchange);
        break;
    case AlmaMessage.Operation.Update:
        userService.updateUser(exchange);
        break;
    case AlmaMessage.Operation.CreateOrUpdate:
        userService.createOrUpdateUser(exchange);
        break;
    case AlmaMessage.Operation.Delete:
        userService.deleteUser(exchange);
        break;
    default:
        throw new UnsupportedOperationException("Operation: " + operation + " not supported");
    }
}

/**
 * Run AlmaUserService.deleteUser(), produces no output.
 * @param exchange the Camel exchange.
 * @throws Exception on errors.
 */
public void deleteUser(final Exchange exchange) throws Exception {
    final Message in = exchange.getIn();
    in.setHeader(AlmaMessage.Header.Status, AlmaMessage.Status.Failed);
    String userId = ExchangeHelper.getMandatoryHeader(exchange, AlmaMessage.Header.UserId, String.class);

    log.debug("Getting user with id {} from ALMA", userId);
    if (userService.deleteUser(userId)) {
        log.debug("User with id {} deleted from Alma.", userId);
        in.setHeader(AlmaMessage.Header.Status, AlmaMessage.Status.Ok);
    } else {
        log.debug("User with id {} NOT deleted from Alma, maybe not found.", userId);
        in.setHeader(AlmaMessage.Header.Status, AlmaMessage.Status.Failed);
    }
}

private OutputStreamBuilder(final Exchange exchange) {
    if (ExchangeHelper.isStreamCachingEnabled(exchange)) {
        outputStream = new CachedOutputStream(exchange);
    } else {
        outputStream = new ByteArrayOutputStream();
    }
}

public boolean process(Exchange exchange, AsyncCallback callback) {
    Iterator<Processor> processors = next().iterator();

    Object lastHandled = exchange.getProperty(Exchange.EXCEPTION_HANDLED);
    exchange.setProperty(Exchange.EXCEPTION_HANDLED, null);

    while (continueRouting(processors, exchange)) {
        exchange.setProperty(Exchange.TRY_ROUTE_BLOCK, true);
        ExchangeHelper.prepareOutToIn(exchange);

        // process the next processor
        Processor processor = processors.next();
        AsyncProcessor async = AsyncProcessorConverterHelper.convert(processor);
        boolean sync = process(exchange, callback, processors, async, lastHandled);

        // continue as long its being processed synchronously
        if (!sync) {
            LOG.trace("Processing exchangeId: {} is continued being processed asynchronously", exchange.getExchangeId());
            // the remainder of the try .. catch .. finally will be completed async
            // so we break out now, then the callback will be invoked which then continue routing from where we left here
            return false;
        }

        LOG.trace("Processing exchangeId: {} is continued being processed synchronously", exchange.getExchangeId());
    }

    ExchangeHelper.prepareOutToIn(exchange);
    exchange.removeProperty(Exchange.TRY_ROUTE_BLOCK);
    exchange.setProperty(Exchange.EXCEPTION_HANDLED, lastHandled);
    LOG.trace("Processing complete for exchangeId: {} >>> {}", exchange.getExchangeId(), exchange);
    callback.done(true);
    return true;
}

protected Endpoint resolveEndpoint(Exchange exchange, Object recipient) {
    // trim strings as end users might have added spaces between separators
    if (recipient instanceof String) {
        recipient = ((String)recipient).trim();
    }
    return ExchangeHelper.resolveEndpoint(exchange, recipient);
}

@Override
public boolean process(Exchange exchange, AsyncCallback callback) {
    try {
        Object newBody = 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();

        // create a new message container so we do not drag specialized message objects along
        // but that is only needed if the old message is a specialized message
        boolean copyNeeded = !(old.getClass().equals(DefaultMessage.class));

        if (copyNeeded) {
            Message msg = new DefaultMessage();
            msg.copyFrom(old);
            msg.setBody(newBody);

            // replace message on exchange
            ExchangeHelper.replaceMessage(exchange, msg, false);
        } else {
            // no copy needed so set replace value directly
            old.setBody(newBody);
        }

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

    callback.done(true);
    return true;
}

public void onComplete(Exchange exchange) {
    if (ExchangeHelper.isFailureHandled(exchange)) {
        // the exchange did not process successfully but was failure handled by the dead letter channel
        // and thus moved to the dead letter queue. We should thus not consider it as complete.
        onFailedMessage(exchange, messageId);
    } else {
        onCompletedMessage(exchange, messageId);
    }
}

protected Endpoint resolveEndpoint(Exchange exchange, Object recipient) {
    // trim strings as end users might have added spaces between separators
    if (recipient instanceof String) {
        recipient = ((String)recipient).trim();
    }
    return ExchangeHelper.resolveEndpoint(exchange, recipient);
}

private void marshalUrlSafe(Exchange exchange, Object graph, OutputStream stream) throws Exception {
    byte[] decoded = ExchangeHelper.convertToMandatoryType(exchange, byte[].class, graph);

    Base64 codec = new Base64(lineLength, lineSeparator, true);
    byte[] encoded = codec.encode(decoded);

    stream.write(encoded);
    stream.flush();
}

private Exchange configureCopyExchange(Exchange exchange) {
    // must use a copy as we dont want it to cause side effects of the original exchange
    Exchange copy = ExchangeHelper.createCorrelatedCopy(exchange, false);
    // set MEP to InOnly as this wire tap is a fire and forget
    copy.setPattern(ExchangePattern.InOnly);
    return copy;
}

@Override
protected void populateInitialHeaders(Map<String, Object> map) {
    if (mailMessage != null) {
        try {
            MailBinding binding = ExchangeHelper.getBinding(getExchange(), MailBinding.class);
            if (binding != null) {
                map.putAll(binding.extractHeadersFromMail(mailMessage, getExchange()));
            }
        } catch (MessagingException e) {
            throw new RuntimeCamelException("Error accessing headers due to: " + e.getMessage(), e);
        }
    }
}

protected Endpoint resolveEndpoint(RoutingSlipIterator iter, Exchange exchange) throws Exception {
    Object nextRecipient = iter.next(exchange);
    Endpoint endpoint = null;
    try {
        endpoint = ExchangeHelper.resolveEndpoint(exchange, nextRecipient);
    } catch (Exception e) {
        if (isIgnoreInvalidEndpoints()) {
            log.info("Endpoint uri is invalid: " + nextRecipient + ". This exception will be ignored.", e);
        } else {
            throw e;
        }
    }
    return endpoint;
}

private static Exchange copyExchangeNoAttachments(Exchange exchange, boolean preserveExchangeId) {
    Exchange answer = ExchangeHelper.createCopy(exchange, preserveExchangeId);
    // we do not want attachments for the splitted sub-messages
    answer.getIn().setAttachments(null);
    // we do not want to copy the message history for splitted sub-messages
    answer.getProperties().remove(Exchange.MESSAGE_HISTORY);
    return answer;
}

/**
 * Creates a new {@link DefaultExchange} instance from the given
 * <code>exchange</code>. The resulting exchange's pattern is defined by
 * <code>pattern</code>.
 *
 * @param source  exchange to copy from.
 * @param pattern exchange pattern to set.
 * @return created exchange.
 */
protected Exchange createResourceExchange(Exchange source, ExchangePattern pattern) {
    // copy exchange, and do not share the unit of work
    Exchange target = ExchangeHelper.createCorrelatedCopy(source, false);
    target.setPattern(pattern);

    // if we share unit of work, we need to prepare the resource exchange
    if (isShareUnitOfWork()) {
        target.setProperty(Exchange.PARENT_UNIT_OF_WORK, source.getUnitOfWork());
        // and then share the unit of work
        target.setUnitOfWork(source.getUnitOfWork());
    }
    return target;
}

public void processDataSet(Exchange originalExchange, DataSet dataSet, int counter) throws Exception {
    Exchange exchange = ExchangeHelper.createCorrelatedCopy(originalExchange, false);
    Message in = exchange.getIn();
    in.setBody(dataSet);
    in.setHeader("CamelFlatpackCounter", counter);
    loadBalancer.process(exchange);
}

protected static Endpoint resolveEndpoint(Exchange exchange, Object recipient) throws NoTypeConversionAvailableException {
    // trim strings as end users might have added spaces between separators
    if (recipient instanceof String) {
        recipient = ((String) recipient).trim();
    } else if (recipient instanceof Endpoint) {
        return (Endpoint) recipient;
    } else {
        // convert to a string type we can work with
        recipient = exchange.getContext().getTypeConverter().mandatoryConvertTo(String.class, exchange, recipient);
    }

    return ExchangeHelper.resolveEndpoint(exchange, recipient);
}

public void processReply(ReplyHolder holder) {
    if (holder != null && isRunAllowed()) {
        try {
            Exchange exchange = holder.getExchange();

            boolean timeout = holder.isTimeout();
            if (timeout) {
                // timeout occurred do a WARN log so its easier to spot in the logs
                if (log.isWarnEnabled()) {
                    log.warn("Timeout occurred after {} millis waiting for reply message with correlationID [{}] on destination {}."
                            + " Setting ExchangeTimedOutException on {} and continue routing.",
                             holder.getRequestTimeout(), holder.getCorrelationId(), replyTo, ExchangeHelper.logIds(exchange));
                }

                // no response, so lets set a timed out exception
                String msg = "reply message with correlationID: " + holder.getCorrelationId() + " not received on destination: " + replyTo;
                exchange.setException(new ExchangeTimedOutException(exchange, holder.getRequestTimeout(), msg));
            } else {

                messageConverter.populateRabbitExchange(exchange, null, holder.getProperties(), holder.getMessage(), true);

                // restore correlation id in case the remote server messed with it
                if (holder.getOriginalCorrelationId() != null) {
                    if (exchange.hasOut()) {
                        exchange.getOut().setHeader(RabbitMQConstants.CORRELATIONID, holder.getOriginalCorrelationId());
                    } else {
                        exchange.getIn().setHeader(RabbitMQConstants.CORRELATIONID, holder.getOriginalCorrelationId());
                    }
                }
            }
        } finally {
            // notify callback
            AsyncCallback callback = holder.getCallback();
            callback.done(false);
        }
    }
}

@Override
protected Object createBody() {
    if (mailMessage != null) {
        MailBinding binding = ExchangeHelper.getBinding(getExchange(), MailBinding.class);
        return binding != null ? binding.extractBodyFromMail(getExchange(), this) : null;
    }
    return null;
}

protected static Endpoint resolveEndpoint(Exchange exchange, Object recipient) {
    // trim strings as end users might have added spaces between separators
    if (recipient instanceof String) {
        recipient = ((String) recipient).trim();
    }
    return ExchangeHelper.resolveEndpoint(exchange, recipient);
}

@Override
public void done(boolean doneSync) {
    try {
        if (exception == null) {
            exchange.removeProperty(Exchange.FAILURE_ENDPOINT);
        } else {
            // set exception back on exchange
            exchange.setException(exception);
            exchange.setProperty(Exchange.EXCEPTION_CAUGHT, exception);
        }
        // set fault flag back
        if (fault) {
            if (exchange.hasOut()) {
                exchange.getOut().setFault(true);
            } else {
                exchange.getIn().setFault(true);
            }
        }

        if (!doneSync) {
            // signal callback to continue routing async
            ExchangeHelper.prepareOutToIn(exchange);
            LOG.trace("Processing complete for exchangeId: {} >>> {}", exchange.getExchangeId(), exchange);
        }
    } finally {
        // callback must always be called
        callback.done(doneSync);
    }
}

/**
 * Performs a defensive copy of the exchange if needed
 *
 * @param exchange the exchange
 * @return the defensive copy, or <tt>null</tt> if not needed (redelivery is not enabled).
 */
protected Exchange defensiveCopyExchangeIfNeeded(Exchange exchange) {
    // only do a defensive copy if redelivery is enabled
    if (redeliveryEnabled) {
        return ExchangeHelper.createCopy(exchange, true);
    } else {
        return null;
    }
}

/**
 * Strategy to determine if the exchange was cancelled or interrupted
 */
protected boolean isCancelledOrInterrupted(Exchange exchange) {
    boolean answer = false;

    if (ExchangeHelper.isInterrupted(exchange)) {
        // mark the exchange to stop continue routing when interrupted
        // as we do not want to continue routing (for example a task has been cancelled)
        exchange.setProperty(Exchange.ROUTE_STOP, Boolean.TRUE);
        answer = true;
    }

    log.trace("Is exchangeId: {} interrupted? {}", exchange.getExchangeId(), answer);
    return answer;
}