@Handler
public String enhanceMessage( String body,  Exchange exchange  ) {
	
	String[] topicParts = exchange.getIn().getHeader("CamelMQTTSubscribeTopic", String.class).split(TOPIC_SEPARTOR);

	if(topicParts.length != 4) {
		throw new IllegalArgumentException("Invalid number of topic components. Expected " + TOPIC_PART_SIZE + ". Was " + topicParts.length);
	}
	
	StringBuilder sb = new StringBuilder();
	sb.append(body);
	sb.append(COMMA);
	sb.append(topicParts[1]);
	sb.append(COMMA);
	sb.append(topicParts[2]);
	sb.append(COMMA);
	sb.append(topicParts[3]);
	
	return sb.toString();
}
 

@Handler
public boolean lastFileProcessed(Exchange exchange) throws Exception {
    FileEntryWorkNote workNote = exchange.getIn().getBody(FileEntryWorkNote.class);

    String batchJobId = workNote.getBatchJobId();
    TenantContext.setJobId(batchJobId);

    if (batchJobDAO.updateFileEntryLatch(workNote.getBatchJobId(), workNote.getFileEntry().getFileName())) {

        RangedWorkNote wn = RangedWorkNote.createSimpleWorkNote(batchJobId);
        exchange.getIn().setBody(wn, RangedWorkNote.class);
        return true;

    }
    
    return false;
}
 

@Handler
public void processCommand(Exchange exch) throws Exception {
    String command = exch.getIn().getBody().toString();
    LOG.info("Received: " + command);
    String[] chunks = command.split("\\|");

    if (JOB_COMPLETED.equals(chunks[0])) {

        LOG.info("Clearing cache at job completion.");


        dumpAspectTrackers(chunks);

    } else {
        LOG.error("Unsupported command");
    }
}
 

@Handler
public void handle(@Body String body, @Headers Map headers, Exchange exchange) {
    Random random = new Random(System.currentTimeMillis());
    if( random.nextBoolean() ) {
        throw new RuntimeCamelException("Random error.. try your luck again next time.");
    }
}
 

@Handler
public void throwNonBpmnException() throws Exception {
  LOG.debug("throwing non bpmn bug");

  switch (getExceptionType()) {
  case NO_EXCEPTION:
    break;
  case NON_BPMN_EXCEPTION:
    throw new Exception("arbitary non bpmn exception");
  case BPMN_EXCEPTION:
    throw new BpmnError("testError");
  }
}
 

@Handler
public void throwNonBpmnException() throws Exception {
  LOG.debug("throwing non bpmn bug");

  switch (getExceptionType()) {
  case NO_EXCEPTION:
    break;
  case NON_BPMN_EXCEPTION:
    throw new Exception("arbitary non bpmn exception");
  case BPMN_EXCEPTION:
    throw new BpmnError("testError");
  }
}
 

@Handler
public List<Out> process(@Body In body) {
    List<Out> out = new ArrayList<>();
    out.add(new Out());

    return out;
}
 

@Handler
public List<Out> process(@Body In body) {
    List<Out> out = new ArrayList<>();
    out.add(new Out());

    return out;
}
 

@Handler
public void log(@Body Object body){
    if(trace) {
        LOGGER.trace("Body is: {}",body);
    } else {
        LOGGER.info("Body is: {}",body);
    }
}
 

@Handler
public String process(String body) {
    if ("error".equals(body)) {
        throw new RuntimeException("Bean Error");
    }
    return body + " World";
}
 

@Handler
public void throwNonBpmnException() throws Exception {
    LOGGER.debug("throwing non bpmn bug");

    switch (getExceptionType()) {
    case NO_EXCEPTION:
        break;
    case NON_BPMN_EXCEPTION:
        throw new Exception("arbitrary non bpmn exception");
    case BPMN_EXCEPTION:
        throw new BpmnError("testError");
    }
}
 

@Handler
public void throwNonBpmnException() throws Exception {
    LOGGER.debug("throwing non bpmn bug");

    switch (getExceptionType()) {
    case NO_EXCEPTION:
        break;
    case NON_BPMN_EXCEPTION:
        throw new Exception("arbitrary non bpmn exception");
    case BPMN_EXCEPTION:
        throw new BpmnError("testError");
    }
}
 

@Handler
public void orderSomeBooks() throws Exception {
    log.info("-------------------------------------------------------------------------------------------------------------------------");
    log.info("Make sure to have Postgres database up and running, as configured in the src/test/resources/META-INF/persistence.xml file");
    log.info("-------------------------------------------------------------------------------------------------------------------------");

    BookOrder order = new BookOrder();
    order.setAmount(1);
    order.setTitle("Camel in Action 2nd ed");

    template.sendBody("jpa:camelinaction.BookOrder", order);

    Thread.sleep(5000);
    log.info("... sleeping for 5 seconds and then stopping the route");

    // now stop the route
    context.stopRoute("books");

    // insert a new order which will sit in the database
    BookOrder order2 = new BookOrder();
    order2.setAmount(3);
    order2.setTitle("ActiveMQ in Action");

    template.sendBody("jpa:camelinaction.BookOrder", order2);

    log.info("-------------------------------------------------------------------------------------------------------------------------");
    log.info("Now we want to provoke a connection error, so stop the postgres database - and then press ENTER to continue!");
    log.info("-------------------------------------------------------------------------------------------------------------------------");

    System.console().readLine();

    context.startRoute("books");
    log.info("... starting route which should indicate some errors, which the bridge error handler should catch and handle");
    log.info("Notice that the consumer will backoff and not poll so fast, instead of every second, it now runs x10 sec.");
    log.info("Press CTRL+C to exit this application!");
}
 

@Handler
public Map getProcessVariables(@Body String body,
                               @Header(Exchange.FILE_NAME) String filename,
                               @Simple("${date:now:yyyy-MM-dd kk:mm:ss}") String timestamp) {
    Map<String, Object> variables = new HashMap<String, Object>();
    variables.put("message", body);
    variables.put("orderid", filename);
    variables.put("timestamp", timestamp);
    return variables;
}
 

@Handler
public String apply(@Body String body) {
    return "Hello " + body;
}
 

@Handler
public int apply(@Body String body) {
    return body.hashCode();
}
 

@Handler
public String[] apply(@Body String body) {
    return new String[]{ "Hiram", "World" };
}
 

@Handler
public void handle(@Body String body) {
    // NO-OP
}
 

@Handler
public void myOverLoadedBean3(String name) {}
 

@Handler
public void decrement(Exchange exchange) {
    LOG.info("Decrementing latch count: " + label);
    latch.countDown();
}