Hi Sushil,
Here is a basic example for you in the Java DSL, you can create and check as
many exchanges as you want. I would say just make sure that you are handling
any exceptions that may happen gracefully. A custom exchange handler is nearly
always needed to handle what to do in event of a crash at some point along the
way. I hope that this helps:
Test.java
public class Test extends SpringRouteBuilder {
private static final transient Log LOG = LogFactory.getLog(Test .class);
public void configure() throws Exception {
CamelContext camelContext = bean(CamelContext.class,"myCamelContext");
ProducerTemplate producerTemplate =
camelContext.createProducerTemplate();
Endpoint amqIncomingEndpoint =
camelContext.getEndpoint("activemq:queue:testqueue.INCOMING");
Endpoint jdbcEndpoint = camelContext.getEndpoint("jdbc:mydatabase");
MyProcessor routeProcessor = new MyProcessor
(producerTemplate,amqIncomingEndpoint,
jdbcEndpoint ,camelContext);
errorHandler(deadLetterChannel().maximumRedeliveries(-1));
from("activemq:queue:testqueue.OUTGOING")
.process(routeProcessor);
}
}
MyProcessor.java
public class MyProcessor implements Processor {
private static final transient Log LOG =
LogFactory.getLog(MyProcessor.class);
private ProducerTemplate producerTemplate;
private Endpoint jdbcEndpoint;
private Endpoint amqIncomingEndpoint;
private CamelContext camelContext;
public MyProcessor (ProducerTemplate producerTemplate,
Endpoint amqIncomingEndpoint,Endpoint jdbcEndpoint,CamelContext camelContext) {
this.producerTemplate = producerTemplate;
this.amqIncomingEndpoint= amqIncomingEndpoint;
this.camelContext = camelContext;
this.jdbcEndpoint = jdbcEndpoint
}
public void process(Exchange exchange) throws Exception {
Map<String, Object> messageHeaders = exchange.getIn().getHeaders();
String sqlStatement = "Some Sql Statement"; // not safe to do it this
way, just for example
// 1 - Create message and check response in this case jdbc
try {
Message myMessage = new DefaultMessage();
myMessage.setBody(sqlStatement);
myMessage.setHeaders(msgHeaders);
Exchange myExchange = new DefaultExchange(camelContext);
myExchange.setIn(myMessage);
Exchange returnInsertDBExg =
producerTemplate.send(jdbcEndpoint,myExchange);
String updateCount =
returnInsertDBExg.getOut().getHeader("jdbc.updateCount", String.class);
if (updateCount.equals("1")) {
LOG.info("Success, 1 row inserted.");
} else if (updateCount.equals("0")) {
LOG.info("Failed, 0 rows inserted.");
} else if (updateCount.equals(null)) {
LOG.debug("Null Row Count");
}
} catch (Exception e) {
LOG.error("Unexpected", e);
}
// 2 - Processes complete, send notification
try {
Message msgNotifyRPM = new DefaultMessage();
msgNotifyRPM.setBody(uploadSuccess, String.class);
msgNotifyRPM.setHeaders(messageHeaders);
Exchange exgNotifyRPM = new DefaultExchange(camelContext);
exgNotifyRPM.setIn(msgNotifyRPM);
Exchange returnExchange = producerTemplate.send(
amqIncomingEndpoint, exgNotifyRPM);
} catch (Exception e) {
LOG.error("Unexpected = ", e);
}
LOG.info("Process Complete.");
}
}
camelContext.xml
<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans";
xmlns:p="http://www.springframework.org/schema/p";
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance";
xmlns:osgi="http://www.springframework.org/schema/osgi";
xsi:schemaLocation="http://www.springframework.org/schema/beans
http://www.springframework.org/schema/beans/spring-beans-2.5.xsd
http://www.springframework.org/schema/osgi
http://www.springframework.org/schema/osgi/spring-osgi.xsd
http://activemq.apache.org/camel/schema/spring
http://activemq.apache.org/camel/schema/spring/camel-spring.xsd";>
<camelContext id="myCamelContext"
xmlns="http://activemq.apache.org/camel/schema/spring";>
<package>com.package.my</package>
</camelContext>
<bean id="mydatabase" class="org.apache.camel.component.jdbc.JdbcComponent">
<property name="dataSource" ref="myjdbc"/>
</bean>
<bean id="myjdbc" class="org.apache.commons.dbcp.BasicDataSource"
destroy-method="close">
<property name="driverClassName" value="${jdbc.driverClassName}" />
<property name="url" value="${jdbc.url}" />
<property name="username" value="${jdbc.username}" />
<property name="password" value="${jdbc.password}" />
</bean>
<bean id="activemq"
class="org.apache.activemq.camel.component.ActiveMQComponent">
<property name="connectionFactory" ref="activeMQVMConnectionFactory" />
<property name="acknowledgementModeName" value="CLIENT_ACKNOWLEDGE" />
<property name="concurrentConsumers" value="1" />
</bean>
</beans>
> Date: Tue, 28 Aug 2012 21:30:14 -0700
> From: sushil_v...@yahoo.co.in
> To: users@camel.apache.org
> Subject: Custom Processor for branching into multiple Sub Exchanges
>
> Hello:
> Instead of using the Splitter, I would like to have a custom processor /
> bean which should be able to sequentially create sub exchanges and send them
> to a child route. It should wait for the execution of the route to finish
> before it sends the next exchange. How best do I implement this?
> Regards,
> Sushil
>
>
>
> --
> View this message in context:
> http://camel.465427.n5.nabble.com/Custom-Processor-for-branching-into-multiple-Sub-Exchanges-tp5718223.html
> Sent from the Camel - Users mailing list archive at Nabble.com.
