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

@SuppressWarnings("unchecked")
public boolean process(Exchange exchange, AsyncCallback callback) {
    Exception cause = exception;

    try {
        if (message != null && type != null) {
            // create the message using simple language so it can be dynamic
            String text = simple.evaluate(exchange, String.class);
            // create a new exception of that type, and provide the message as
            Constructor<?> constructor = type.getDeclaredConstructor(String.class);
            cause = (Exception) constructor.newInstance(text);
            exchange.setException(cause);
        } else {
            exchange.setException(cause);
        }
    } catch (Throwable e) {
        exchange.setException(new CamelExchangeException("Error creating new instance of " + exception.getClass(), exchange, e));
    }

    callback.done(true);
    return true;
}

public void handleException(String message, Exchange exchange, Throwable exception) {
    try {
        if (!isSuppressLogging()) {
            String msg = CamelExchangeException.createExceptionMessage(message, exchange, exception);
            if (isCausedByRollbackExchangeException(exception)) {
                // do not log stack trace for intended rollbacks
                logger.log(msg);
            } else {
                if (exception != null) {
                    logger.log(msg, exception);
                } else {
                    logger.log(msg);
                }
            }
        }
    } catch (Throwable e) {
        // the logging exception handler must not cause new exceptions to occur
    }
}

public void testSplitWithTryCatchAndRollbackException() throws Exception {
    MockEndpoint split = getMockEndpoint("mock:split");
    MockEndpoint ile = getMockEndpoint("mock:ile");
    MockEndpoint exception = getMockEndpoint("mock:exception");

    split.expectedBodiesReceived("A", "B");
    ile.expectedMessageCount(0);
    exception.expectedMessageCount(1);

    try {
        template.sendBody("direct:start", "A,B,Kaboom,C");
        fail("Should thrown an exception");
    } catch (CamelExecutionException e) {
        CamelExchangeException ee = assertIsInstanceOf(CamelExchangeException.class, e.getCause());
        assertTrue(ee.getMessage().startsWith("Sequential processing failed for number 2."));
        RollbackExchangeException re = assertIsInstanceOf(RollbackExchangeException.class, ee.getCause());
        assertTrue(re.getMessage().startsWith("Intended rollback"));
    }

    assertMockEndpointsSatisfied();
}

public void testSplitWithTryCatchAndRollbacILEAndException() throws Exception {
    MockEndpoint split = getMockEndpoint("mock:split");
    MockEndpoint ile = getMockEndpoint("mock:ile");
    MockEndpoint exception = getMockEndpoint("mock:exception");

    split.expectedBodiesReceived("A", "B");
    ile.expectedMessageCount(1);
    exception.expectedMessageCount(1);

    try {
        template.sendBody("direct:start", "A,Forced,B,Kaboom,C");
        fail("Should thrown an exception");
    } catch (CamelExecutionException e) {
        CamelExchangeException ee = assertIsInstanceOf(CamelExchangeException.class, e.getCause());
        assertTrue(ee.getMessage().startsWith("Sequential processing failed for number 3."));
        RollbackExchangeException re = assertIsInstanceOf(RollbackExchangeException.class, ee.getCause());
        assertTrue(re.getMessage().startsWith("Intended rollback"));
    }

    assertMockEndpointsSatisfied();
}

@Test
public void testInOutWithRequestBody() throws Exception {
    MockEndpoint mock = getMockEndpoint("mock:pickedUp");
    mock.expectedMessageCount(1);
    // this direct endpoint should receive an exception
    try {
        Future<Object> obj = template.asyncRequestBody("direct:in", "Hello World");
        // wait five seconds at most; else, let's assume something went wrong
        obj.get(5000, TimeUnit.MILLISECONDS);
    } catch (Exception e) {
        // if we receive an exception, the async routing engine is working correctly
        // before the Enricher was fixed for cases where routing was async and the AggregationStrategy 
        // threw an exception, the call to requestBody would stall indefinitely
        // unwrap the exception chain
        assertTrue(e instanceof ExecutionException);
        assertTrue(e.getCause() instanceof CamelExecutionException);
        assertTrue(e.getCause().getCause() instanceof CamelExchangeException);
        assertTrue(e.getCause().getCause().getCause() instanceof RuntimeException);
        assertTrue(e.getCause().getCause().getCause().getMessage().equals("Bang! Unhandled exception"));
        mock.assertIsSatisfied();
        return;
    }
    fail("Expected an RuntimeException");
}

public void testMulticastParalllelStopOnExceptionStop() throws Exception {
    // we run in parallel so we may get 0 or 1 messages
    getMockEndpoint("mock:foo").expectedMinimumMessageCount(0);
    getMockEndpoint("mock:bar").expectedMinimumMessageCount(0);
    getMockEndpoint("mock:baz").expectedMinimumMessageCount(0);
    // we should not complete and thus 0
    getMockEndpoint("mock:result").expectedMessageCount(0);

    try {
        template.sendBody("direct:start", "Kaboom");
        fail("Should thrown an exception");
    } catch (CamelExecutionException e) {
        CamelExchangeException cause = assertIsInstanceOf(CamelExchangeException.class, e.getCause());
        assertTrue(cause.getMessage().startsWith("Parallel processing failed for number "));
        assertEquals("Forced", cause.getCause().getMessage());

        String body = cause.getExchange().getIn().getBody(String.class);
        assertTrue(body.contains("Kaboom"));
    }

    assertMockEndpointsSatisfied();
}

public void testSplitStopOnExceptionStop() throws Exception {
    MockEndpoint mock = getMockEndpoint("mock:split");
    // we do stop so we stop splitting when the exception occurs and thus we only receive 1 message
    mock.expectedBodiesReceived("Hello World");

    try {
        template.sendBody("direct:start", "Hello World,Kaboom,Bye World");
        fail("Should thrown an exception");
    } catch (CamelExecutionException e) {
        CamelExchangeException cause = assertIsInstanceOf(CamelExchangeException.class, e.getCause());
        assertTrue(cause.getMessage().startsWith("Sequential processing failed for number 1."));
        assertEquals("Forced", cause.getCause().getMessage());
    }

    assertMockEndpointsSatisfied();
}

public void testMulticastStopOnExceptionStop() throws Exception {
    getMockEndpoint("mock:foo").expectedBodiesReceived("Kaboom");
    getMockEndpoint("mock:bar").expectedMessageCount(0);
    // we do stop so we should NOT continue and thus baz do not receive any message
    getMockEndpoint("mock:baz").expectedMessageCount(0);
    getMockEndpoint("mock:result").expectedMessageCount(0);

    try {
        template.sendBody("direct:start", "Kaboom");
        fail("Should thrown an exception");
    } catch (CamelExecutionException e) {
        CamelExchangeException cause = assertIsInstanceOf(CamelExchangeException.class, e.getCause());
        assertTrue(cause.getMessage().startsWith("Sequential processing failed for number 1."));
        assertEquals("Forced", cause.getCause().getMessage());
    }

    assertMockEndpointsSatisfied();
}

public void testRecipientListStopOnException() throws Exception {
    getMockEndpoint("mock:result").expectedMessageCount(0);
    getMockEndpoint("mock:a").expectedMessageCount(1);
    getMockEndpoint("mock:b").expectedMessageCount(1);
    getMockEndpoint("mock:c").expectedMessageCount(0);

    try {
        template.sendBodyAndHeader("direct:start", "Hello World", "foo", "direct:a,direct:b,direct:c");
        fail("Should have thrown exception");
    } catch (CamelExecutionException e) {
        assertIsInstanceOf(CamelExchangeException.class, e.getCause());
        assertIsInstanceOf(IllegalArgumentException.class, e.getCause().getCause());
        assertEquals("Damn", e.getCause().getCause().getMessage());
    }

    assertMockEndpointsSatisfied();
}

public void testAsyncEndpoint() throws Exception {
    getMockEndpoint("mock:error").expectedMessageCount(0);
    getMockEndpoint("mock:result").expectedMessageCount(0);

    try {
        template.requestBody("direct:start", "Hello Camel", String.class);
        fail("Should throw exception");
    } catch (CamelExecutionException e) {
        CamelExchangeException cause = assertIsInstanceOf(CamelExchangeException.class, e.getCause());
        assertTrue(cause.getMessage().startsWith("Simulated error at attempt 1."));
    }

    assertMockEndpointsSatisfied();

    assertFalse("Should use different threads", beforeThreadName.equalsIgnoreCase(afterThreadName));
}

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

    JobParameters jobParameters = prepareJobParameters(exchange.getIn().getHeaders());
    String messageJobName = jobParameters.getString(SpringBatchConstants.JOB_NAME);

    Job job2run = this.job;

    if (messageJobName != null) {
        job2run = CamelContextHelper.mandatoryLookup(getEndpoint().getCamelContext(), messageJobName, Job.class);
    }

    if (job2run == null) {
        exchange.setException(new CamelExchangeException("jobName was not specified in the endpoint construction "
                + " and header " + SpringBatchConstants.JOB_NAME + " could not be found", exchange));
        return;
    }

    JobExecution jobExecution = jobLauncher.run(job2run, jobParameters);
    exchange.getOut().getHeaders().putAll(exchange.getIn().getHeaders());
    exchange.getOut().setBody(jobExecution);
}

public boolean process(Exchange exchange, final AsyncCallback callback) {
    boolean flag = true;

    if ((((RouteboxDirectEndpoint)getRouteboxEndpoint()).getConsumer() == null) 
        && ((getRouteboxEndpoint()).getConfig().isSendToConsumer())) {
        exchange.setException(new CamelExchangeException("No consumers available on endpoint: " + getRouteboxEndpoint(), exchange));
        callback.done(true);
        flag = true;
    } else {
        try {
            LOG.debug("Dispatching to Inner Route {}", exchange);

            RouteboxDispatcher dispatcher = new RouteboxDispatcher(producer);
            exchange = dispatcher.dispatchAsync(getRouteboxEndpoint(), exchange);      
            if (getRouteboxEndpoint().getConfig().isSendToConsumer()) {
                AsyncProcessor processor = ((RouteboxDirectEndpoint)getRouteboxEndpoint()).getConsumer().getAsyncProcessor();
                flag = processor.process(exchange, callback);
            } 
        } catch (Exception e) {
            getExceptionHandler().handleException("Error processing exchange", exchange, e);
        }
    }
    return flag;
}

protected void doAddEmbeddedCartridge(Exchange exchange, IDomain domain) throws CamelExchangeException {
    String name = exchange.getIn().getHeader(OpenShiftConstants.APPLICATION, getEndpoint().getApplication(), String.class);
    if (name == null) {
        throw new CamelExchangeException("Application not specified", exchange);
    }

    IApplication app = domain.getApplicationByName(name);
    if (app == null) {
        throw new CamelExchangeException("Application with id " + name + " not found.", exchange);
    } else {
        String embeddedCartridgeName = exchange.getIn().getHeader(OpenShiftConstants.EMBEDDED_CARTRIDGE_NAME, getEndpoint().getApplication(), String.class);
        if (ObjectHelper.isNotEmpty(embeddedCartridgeName)) {
            IEmbeddedCartridge p = app.addEmbeddableCartridge((new LatestEmbeddableCartridge(embeddedCartridgeName)).get(app));
            exchange.getIn().setBody(p.getDisplayName());
        } else {
            throw new CamelExchangeException("Cartridge not specified", exchange);
        }
    }
}

protected void doRemoveEmbeddedCartridge(Exchange exchange, IDomain domain) throws CamelExchangeException {
    String name = exchange.getIn().getHeader(OpenShiftConstants.APPLICATION, getEndpoint().getApplication(), String.class);
    if (name == null) {
        throw new CamelExchangeException("Application not specified", exchange);
    }

    IApplication app = domain.getApplicationByName(name);
    if (app == null) {
        throw new CamelExchangeException("Application with id " + name + " not found.", exchange);
    } else {
        String embeddedCartridgeName = exchange.getIn().getHeader(OpenShiftConstants.EMBEDDED_CARTRIDGE_NAME, getEndpoint().getApplication(), String.class);
        if (ObjectHelper.isNotEmpty(embeddedCartridgeName)) {
            IEmbeddableCartridge removingCartridge = (new LatestEmbeddableCartridge(embeddedCartridgeName)).get(app);
            for (IEmbeddedCartridge cartridge : app.getEmbeddedCartridges()) {
                if (cartridge.equals(removingCartridge)) {
                    cartridge.destroy();
                    exchange.getIn().setBody(cartridge.getDisplayName());
                }
            }
        } else {
            throw new CamelExchangeException("Cartridge not specified", exchange);
        }
    }
}

protected void doScaleUp(Exchange exchange, IDomain domain) throws CamelExchangeException {
    String name = exchange.getIn().getHeader(OpenShiftConstants.APPLICATION, getEndpoint().getApplication(), String.class);
    if (name == null) {
        throw new CamelExchangeException("Application not specified", exchange);
    }

    IApplication app = domain.getApplicationByName(name);
    if (app == null) {
        throw new CamelExchangeException("Application with id " + name + " not found.", exchange);
    } else {
        try {
            app.scaleUp();
            ApplicationScale result = app.getApplicationScale();
            exchange.getIn().setBody(result.getValue());
        } catch (OpenShiftException e) {
            throw new CamelExchangeException("Application with id " + name + " is not scalable", exchange);
        }
    }
}

protected void doScaleDown(Exchange exchange, IDomain domain) throws CamelExchangeException {
    String name = exchange.getIn().getHeader(OpenShiftConstants.APPLICATION, getEndpoint().getApplication(), String.class);
    if (name == null) {
        throw new CamelExchangeException("Application not specified", exchange);
    }

    IApplication app = domain.getApplicationByName(name);
    if (app == null) {
        throw new CamelExchangeException("Application with id " + name + " not found.", exchange);
    } else {
        ApplicationScale scale = app.getApplicationScale();
        if (scale.getValue().equals(ApplicationScale.NO_SCALE.getValue())) {
            log.info("Scaling on application with id " + name + " is not enabled");
        } else {
            app.scaleDown();
            ApplicationScale result = app.getApplicationScale();
            exchange.getIn().setBody(result.getValue());
        }
    }
}

protected void doSetDeploymentType(Exchange exchange, IDomain domain) throws CamelExchangeException {
    String name = exchange.getIn().getHeader(OpenShiftConstants.APPLICATION, getEndpoint().getApplication(), String.class);
    if (name == null) {
        throw new CamelExchangeException("Application not specified", exchange);
    }

    IApplication app = domain.getApplicationByName(name);
    if (app == null) {
        throw new CamelExchangeException("Application with id " + name + " not found.", exchange);
    } else {
        String deploymentType = exchange.getIn().getHeader(OpenShiftConstants.DEPLOYMENT_TYPE, getEndpoint().getApplication(), String.class);
        if (ObjectHelper.isNotEmpty(deploymentType)) {
            String result = app.setDeploymentType(deploymentType);
            exchange.getIn().setBody(result);
        } else {
            throw new CamelExchangeException("Deployment Type not specified", exchange);
        }
    }
}

protected void doAddEnvironmentVariable(Exchange exchange, IDomain domain) throws CamelExchangeException {
    String name = exchange.getIn().getHeader(OpenShiftConstants.APPLICATION, getEndpoint().getApplication(), String.class);
    if (name == null) {
        throw new CamelExchangeException("Application not specified", exchange);
    }

    IApplication app = domain.getApplicationByName(name);
    if (app == null) {
        throw new CamelExchangeException("Application with id " + name + " not found.", exchange);
    } else {
        String variableName = exchange.getIn().getHeader(OpenShiftConstants.ENVIRONMENT_VARIABLE_NAME, getEndpoint().getApplication(), String.class);
        String variableValue = exchange.getIn().getHeader(OpenShiftConstants.ENVIRONMENT_VARIABLE_VALUE, getEndpoint().getApplication(), String.class);
        if (!app.canUpdateEnvironmentVariables()) {
            throw new CamelExchangeException("The application with id " + name + " can't update Environment Variables", exchange);
        }
        if (ObjectHelper.isNotEmpty(variableName) && ObjectHelper.isNotEmpty(variableValue)) {
            IEnvironmentVariable result = app.addEnvironmentVariable(variableName, variableValue);
            exchange.getIn().setBody(result.getName());
        } else {
            throw new CamelExchangeException("Environment variable not correctly specified", exchange);
        }
    }
}

protected void doAddMultipleEnvironmentVariables(Exchange exchange, IDomain domain) throws CamelExchangeException {
    String name = exchange.getIn().getHeader(OpenShiftConstants.APPLICATION, getEndpoint().getApplication(), String.class);
    if (name == null) {
        throw new CamelExchangeException("Application not specified", exchange);
    }

    IApplication app = domain.getApplicationByName(name);
    if (app == null) {
        throw new CamelExchangeException("Application with id " + name + " not found.", exchange);
    } else {
        Map environmentVariables = exchange.getIn().getHeader(OpenShiftConstants.ENVIRONMENT_VARIABLE_MAP, getEndpoint().getApplication(), Map.class);
        if (!app.canUpdateEnvironmentVariables()) {
            throw new CamelExchangeException("The application with id " + name + " can't update Environment Variables", exchange);
        }
        if (ObjectHelper.isNotEmpty(environmentVariables)) {
            Map<String, IEnvironmentVariable> result = app.addEnvironmentVariables(environmentVariables);
            exchange.getIn().setBody(result);
        } else {
            throw new CamelExchangeException("Environment variables not correctly specified", exchange);
        }
    }
}

protected void doUpdateEnvironmentVariable(Exchange exchange, IDomain domain) throws CamelExchangeException {
    String name = exchange.getIn().getHeader(OpenShiftConstants.APPLICATION, getEndpoint().getApplication(), String.class);
    if (name == null) {
        throw new CamelExchangeException("Application not specified", exchange);
    }

    IApplication app = domain.getApplicationByName(name);
    if (app == null) {
        throw new CamelExchangeException("Application with id " + name + " not found.", exchange);
    } else {
        String variableName = exchange.getIn().getHeader(OpenShiftConstants.ENVIRONMENT_VARIABLE_NAME, getEndpoint().getApplication(), String.class);
        String variableValue = exchange.getIn().getHeader(OpenShiftConstants.ENVIRONMENT_VARIABLE_VALUE, getEndpoint().getApplication(), String.class);
        if (!app.canUpdateEnvironmentVariables()) {
            throw new CamelExchangeException("The application with id " + name + " can't update Environment Variables", exchange);
        }
        if (ObjectHelper.isNotEmpty(variableName) && ObjectHelper.isNotEmpty(variableValue)) {
            IEnvironmentVariable result = app.updateEnvironmentVariable(variableName, variableValue);
            exchange.getIn().setBody(result.getName());
        } else {
            throw new CamelExchangeException("Environment variable not correctly specified", exchange);
        }
    }
}

protected void doGetEnvironmentVariableValue(Exchange exchange, IDomain domain) throws CamelExchangeException {
    String name = exchange.getIn().getHeader(OpenShiftConstants.APPLICATION, getEndpoint().getApplication(), String.class);
    if (name == null) {
        throw new CamelExchangeException("Application not specified", exchange);
    }

    IApplication app = domain.getApplicationByName(name);
    if (app == null) {
        throw new CamelExchangeException("Application with id " + name + " not found.", exchange);
    } else {
        String variableName = exchange.getIn().getHeader(OpenShiftConstants.ENVIRONMENT_VARIABLE_NAME, getEndpoint().getApplication(), String.class);
        if (!app.canGetEnvironmentVariables()) {
            throw new CamelExchangeException("The application with id " + name + " can't get Environment Variables", exchange);
        }
        if (ObjectHelper.isNotEmpty(variableName)) {
            IEnvironmentVariable result = app.getEnvironmentVariable(variableName);
            exchange.getIn().setBody(result.getValue());
        } else {
            throw new CamelExchangeException("Environment variable name not specified", exchange);
        }
    }
}

protected void doRemoveEnvironmentVariable(Exchange exchange, IDomain domain) throws CamelExchangeException {
    String name = exchange.getIn().getHeader(OpenShiftConstants.APPLICATION, getEndpoint().getApplication(), String.class);
    if (name == null) {
        throw new CamelExchangeException("Application not specified", exchange);
    }

    IApplication app = domain.getApplicationByName(name);
    if (app == null) {
        throw new CamelExchangeException("Application with id " + name + " not found.", exchange);
    } else {
        String variableName = exchange.getIn().getHeader(OpenShiftConstants.ENVIRONMENT_VARIABLE_NAME, getEndpoint().getApplication(), String.class);
        if (!app.canGetEnvironmentVariables()) {
            throw new CamelExchangeException("The application with id " + name + " can't get Environment Variables", exchange);
        }
        if (ObjectHelper.isNotEmpty(variableName)) {
            app.removeEnvironmentVariable(variableName);
            exchange.getIn().setBody(variableName);
        } else {
            throw new CamelExchangeException("Environment variable name not specified", exchange);
        }
    }
}

protected void doAddAlias(Exchange exchange, IDomain domain) throws CamelExchangeException {
    String name = exchange.getIn().getHeader(OpenShiftConstants.APPLICATION, getEndpoint().getApplication(), String.class);
    if (name == null) {
        throw new CamelExchangeException("Application not specified", exchange);
    }

    IApplication app = domain.getApplicationByName(name);
    if (app == null) {
        throw new CamelExchangeException("Application with id " + name + " not found.", exchange);
    } else {
        String alias = exchange.getIn().getHeader(OpenShiftConstants.APPLICATION_ALIAS, getEndpoint().getApplication(), String.class);
        if (!app.canGetEnvironmentVariables()) {
            throw new CamelExchangeException("The application with id " + name + " can't get Environment Variables", exchange);
        }
        if (ObjectHelper.isNotEmpty(alias)) {
            app.addAlias(alias);
            exchange.getIn().setBody(alias);
        } else {
            throw new CamelExchangeException("Application Alias name not specified", exchange);
        }
    }
}

protected void doRemoveAlias(Exchange exchange, IDomain domain) throws CamelExchangeException {
    String name = exchange.getIn().getHeader(OpenShiftConstants.APPLICATION, getEndpoint().getApplication(), String.class);
    if (name == null) {
        throw new CamelExchangeException("Application not specified", exchange);
    }

    IApplication app = domain.getApplicationByName(name);
    if (app == null) {
        throw new CamelExchangeException("Application with id " + name + " not found.", exchange);
    } else {
        String alias = exchange.getIn().getHeader(OpenShiftConstants.APPLICATION_ALIAS, getEndpoint().getApplication(), String.class);
        if (!app.canGetEnvironmentVariables()) {
            throw new CamelExchangeException("The application with id " + name + " can't get Environment Variables", exchange);
        }
        if (ObjectHelper.isNotEmpty(alias)) {
            app.removeAlias(alias);
            exchange.getIn().setBody(alias);
        } else {
            throw new CamelExchangeException("Application Alias not specified", exchange);
        }
    }
}

@Test
public void testHttpGetWithParamsViaURI() throws Exception {
    // these tests does not run well on Windows
    if (isPlatform("windows")) {
        return;
    }

    MockEndpoint mock = getMockEndpoint("mock:result");
    mock.expectedMessageCount(0);

    // give Jetty time to startup properly
    Thread.sleep(1000);

    // use another port with no connection
    try {
        template.requestBody("jetty://http://localhost:9999/myservice", null, Object.class);
        fail("Should have thrown an exception");
    } catch (Exception e) {
        CamelExchangeException cause = assertIsInstanceOf(CamelExchangeException.class, e.getCause());
        assertIsInstanceOf(IOException.class, cause.getCause());
    }

    assertMockEndpointsSatisfied();
}

public boolean process(final Exchange exchange, final AsyncCallback callback) {
    executor.submit(new Callable<Object>() {
        public Object call() throws Exception {
            LOG.info("Simulating a task which takes " + getEndpoint().getDelay() + " millis to reply");
            Thread.sleep(getEndpoint().getDelay());

            int count = counter.incrementAndGet();
            if (getEndpoint().getFailFirstAttempts() >= count) {
                LOG.info("Simulating a failure at attempt " + count);
                exchange.setException(new CamelExchangeException("Simulated error at attempt " + count, exchange));
            } else {
                String reply = getEndpoint().getReply();
                exchange.getOut().setBody(reply);
                LOG.info("Setting reply " + reply);
            }

            LOG.info("Callback done(false)");
            callback.done(false);
            return null;
        }
    });

    // indicate from this point forward its being routed asynchronously
    LOG.info("Task submitted, now tell Camel routing engine to that this Exchange is being continued asynchronously");
    return false;
}

/**
 * Asynchronously writes the given body to MINA session. Will wait at most for
 * 10 seconds until the body has been written.
 *
 * @param session  the MINA session
 * @param body     the body to write (send)
 * @param exchange the exchange
 * @throws CamelExchangeException is thrown if the body could not be written for some reasons
 *                                (eg remote connection is closed etc.)
 */
public static void writeBody(IoSession session, Object body, Exchange exchange) throws CamelExchangeException {
    // the write operation is asynchronous. Use WriteFuture to wait until the session has been written
    WriteFuture future = session.write(body);
    // must use a timeout (we use 10s) as in some very high performance scenarios a write can cause 
    // thread hanging forever
    LOG.trace("Waiting for write to complete for body: {} using session: {}", body, session);
    if (!future.awaitUninterruptibly(10000L)) {
        String message = "Cannot write body: " + body.getClass().getCanonicalName() + " using session: " + session;
        if (future.getException() != null) {
            throw new CamelExchangeException(message, exchange, future.getException());
        } else {
            throw new CamelExchangeException(message, exchange);
        }
    }
}

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

    try {
        SshResult result = SshHelper.sendExecCommand(command, endpoint, client);
        exchange.getOut().setBody(result.getStdout());
        exchange.getOut().setHeader(SshResult.EXIT_VALUE, result.getExitValue());
        exchange.getOut().setHeader(SshResult.STDERR, result.getStderr());
    } catch (Exception e) {
        throw new CamelExchangeException("Cannot execute command: " + command, exchange, e);
    }

    // propagate headers and attachments
    exchange.getOut().getHeaders().putAll(in.getHeaders());
    exchange.getOut().setAttachments(in.getAttachments());
}

protected void doTaskCompleted(Throwable ex) {
    if (ex instanceof TimeoutException) {
        exchange.setException(new ExchangeTimedOutException(exchange, request.getTimeout()));
    } else {
        exchange.setException(new CamelExchangeException("JettyClient failed cause by: " + ex.getMessage(), exchange, ex));
    }
    done.countDown();

    if (callback != null) {
        // now invoke callback to indicate we are done async
        callback.done(false);
    }
}

private void handleDeliverSM(Exchange exchange) throws CamelExchangeException {
    handleSmscAddress(exchange,
        SmppConstants.SOURCE_ADDR, SmppConstants.SOURCE_ADDR_TON,
        originHeaderName);

    handleSmscAddress(exchange,
        SmppConstants.DEST_ADDR, SmppConstants.DEST_ADDR_TON,
        destinationHeaderName);
}

/**
 * Enqueues an exchange for later batch processing.
 */
public boolean process(Exchange exchange, AsyncCallback callback) {
    try {
        // if batch consumer is enabled then we need to adjust the batch size
        // with the size from the batch consumer
        if (isBatchConsumer()) {
            int size = exchange.getProperty(Exchange.BATCH_SIZE, Integer.class);
            if (batchSize != size) {
                batchSize = size;
                LOG.trace("Using batch consumer completion, so setting batch size to: {}", batchSize);
            }
        }

        // validate that the exchange can be used
        if (!isValid(exchange)) {
            if (isIgnoreInvalidExchanges()) {
                LOG.debug("Invalid Exchange. This Exchange will be ignored: {}", exchange);
            } else {
                throw new CamelExchangeException("Exchange is not valid to be used by the BatchProcessor", exchange);
            }
        } else {
            // exchange is valid so enqueue the exchange
            sender.enqueueExchange(exchange);
        }
    } catch (Throwable e) {
        exchange.setException(e);
    }
    callback.done(true);
    return true;
}

@Override
public void run() {
    try {
        if (parallelAggregate) {
            doAggregateInternal(getAggregationStrategy(subExchange), result, subExchange);
        } else {
            doAggregate(getAggregationStrategy(subExchange), result, subExchange);
        }
    } catch (Throwable e) {
        // wrap in exception to explain where it failed
        subExchange.setException(new CamelExchangeException("Parallel processing failed for number " + aggregated.get(), subExchange, e));
    } finally {
        aggregated.incrementAndGet();
    }
}

/**
 * This method is invoked when a Quartz job is fired.
 *
 * @param jobExecutionContext the Quartz Job context
 */
public void onJobExecute(final JobExecutionContext jobExecutionContext) throws JobExecutionException {
    boolean run = true;
    LoadBalancer balancer = getLoadBalancer();
    if (balancer instanceof ServiceSupport) {
        run = ((ServiceSupport) balancer).isRunAllowed();
    }

    if (!run) {
        // quartz scheduler could potential trigger during a route has been shutdown
        LOG.warn("Cannot execute Quartz Job with context: " + jobExecutionContext + " because processor is not started: " + balancer);
        return;
    }

    LOG.debug("Firing Quartz Job with context: {}", jobExecutionContext);
    Exchange exchange = createExchange(jobExecutionContext);
    try {
        balancer.process(exchange);

        if (exchange.getException() != null) {
            // propagate the exception back to Quartz
            throw new JobExecutionException(exchange.getException());
        }
    } catch (Exception e) {
        // log the error
        LOG.error(CamelExchangeException.createExceptionMessage("Error processing exchange", exchange, e));

        // and rethrow to let quartz handle it
        if (e instanceof JobExecutionException) {
            throw (JobExecutionException) e;
        }
        throw new JobExecutionException(e);
    }
}

private Expression createIncExpression(final Expression leftExp) {
    return new Expression() {
        @Override
        public <T> T evaluate(Exchange exchange, Class<T> type) {
            Number num = leftExp.evaluate(exchange, Number.class);
            if (num != null) {
                long val = num.longValue();
                val++;

                // convert value back to same type as input as we want to preserve type
                Object left = leftExp.evaluate(exchange, Object.class);
                try {
                    left = exchange.getContext().getTypeConverter().mandatoryConvertTo(left.getClass(), exchange, val);
                } catch (NoTypeConversionAvailableException e) {
                    throw ObjectHelper.wrapRuntimeCamelException(e);
                }

                // and return the result
                return exchange.getContext().getTypeConverter().convertTo(type, left);
            }
            // cannot convert the expression as a number
            Exception cause = new CamelExchangeException("Cannot evaluate " + leftExp + " as a number", exchange);
            throw ObjectHelper.wrapRuntimeCamelException(cause);
        }

        @Override
        public String toString() {
            return left + operator.toString();
        }
    };
}

private Expression createDecExpression(final Expression leftExp) {
    return new Expression() {
        @Override
        public <T> T evaluate(Exchange exchange, Class<T> type) {
            Number num = leftExp.evaluate(exchange, Number.class);
            if (num != null) {
                long val = num.longValue();
                val--;

                // convert value back to same type as input as we want to preserve type
                Object left = leftExp.evaluate(exchange, Object.class);
                try {
                    left = exchange.getContext().getTypeConverter().mandatoryConvertTo(left.getClass(), exchange, val);
                } catch (NoTypeConversionAvailableException e) {
                    throw ObjectHelper.wrapRuntimeCamelException(e);
                }

                // and return the result
                return exchange.getContext().getTypeConverter().convertTo(type, left);
            }
            // cannot convert the expression as a number
            Exception cause = new CamelExchangeException("Cannot evaluate " + leftExp + " as a number", exchange);
            throw ObjectHelper.wrapRuntimeCamelException(cause);
        }

        @Override
        public String toString() {
            return left + operator.toString();
        }
    };
}

protected Object afterInvoke(Method method, Exchange exchange, ExchangePattern pattern, boolean isFuture) throws Exception {
    // check if we had an exception
    Throwable cause = exchange.getException();
    if (cause != null) {
        Throwable found = findSuitableException(cause, method);
        if (found != null) {
            if (found instanceof Exception) {
                throw (Exception)found;
            } else {
                // wrap as exception
                throw new CamelExchangeException("Error processing exchange", exchange, cause);
            }
        }
        // special for runtime camel exceptions as they can be nested
        if (cause instanceof RuntimeCamelException) {
            // if the inner cause is a runtime exception we can throw it
            // directly
            if (cause.getCause() instanceof RuntimeException) {
                throw (RuntimeException)((RuntimeCamelException)cause).getCause();
            }
            throw (RuntimeCamelException)cause;
        }
        // okay just throw the exception as is
        if (cause instanceof Exception) {
            throw (Exception)cause;
        } else {
            // wrap as exception
            throw new CamelExchangeException("Error processing exchange", exchange, cause);
        }
    }

    Class<?> to = isFuture ? getGenericType(exchange.getContext(), method.getGenericReturnType()) : method.getReturnType();

    // do not return a reply if the method is VOID
    if (to == Void.TYPE) {
        return null;
    }

    return getBody(exchange, to);
}

public void process(Exchange exchange) throws Exception {
    assertEquals("Hello World", exchange.getIn().getBody(String.class));
    if (camelException) {
        throw new CamelExchangeException("I cannot do it", exchange);
    } else {
        throw new IllegalArgumentException("Forced");
    }
}

protected RouteBuilder createRouteBuilder() throws Exception {
    return new RouteBuilder() {
        // START SNIPPET e1
        public void configure() throws Exception {
            // configure the error handler to use my policy instead of the default from Camel
            errorHandler(deadLetterChannel("mock:error").exceptionPolicyStrategy(new MyPolicy()));

            onException(MyPolicyException.class)
                .maximumRedeliveries(1)
                .setHeader(MESSAGE_INFO, constant("Damm my policy exception"))
                .to(ERROR_QUEUE);

            onException(CamelException.class)
                .maximumRedeliveries(3)
                .setHeader(MESSAGE_INFO, constant("Damm a Camel exception"))
                .to(ERROR_QUEUE);
            // END SNIPPET e1

            from("direct:a").process(new Processor() {
                public void process(Exchange exchange) throws Exception {
                    String s = exchange.getIn().getBody(String.class);
                    if ("Hello Camel".equals(s)) {
                        throw new CamelExchangeException("Forced for testing", exchange);
                    }
                    exchange.getOut().setBody("Hello World");
                }
            }).to("mock:result");
        }
    };
}

private void setupPolicies() {
    strategy = new DefaultExceptionPolicyStrategy();
    policies = new HashMap<ExceptionPolicyKey, OnExceptionDefinition>();
    type1 = new OnExceptionDefinition(CamelExchangeException.class);
    type2 = new OnExceptionDefinition(Exception.class);
    type3 = new OnExceptionDefinition(IOException.class);
    policies.put(new ExceptionPolicyKey(null, CamelExchangeException.class, null), type1);
    policies.put(new ExceptionPolicyKey(null, Exception.class, null), type2);
    policies.put(new ExceptionPolicyKey(null, IOException.class, null), type3);
}

private void setupPoliciesNoTopLevelException() {
    // without the top level exception that can be used as fallback
    strategy = new DefaultExceptionPolicyStrategy();
    policies = new HashMap<ExceptionPolicyKey, OnExceptionDefinition>();
    type1 = new OnExceptionDefinition(CamelExchangeException.class);
    type3 = new OnExceptionDefinition(IOException.class);
    policies.put(new ExceptionPolicyKey(null, CamelExchangeException.class, null), type1);
    policies.put(new ExceptionPolicyKey(null, IOException.class, null), type3);
}