Apache Camel with SQL Example

This example shows how to exchange data using a shared database table.
The example has two Camel routes. The first route insert new data into the table, triggered by a timer to run every 5th second.
The second route pickup the newly inserted rows from the table, process the row(s), and mark the row(s) as processed when done; to avoid picking up the same rows again..

package com.javavillage.camel.sql;

import java.util.Date;
import java.util.HashMap;
import java.util.Map;
import java.util.Random;

 * Bean that generates and process NewTopic.
public class NewTopicBean {

    private Random ran = new Random();

     * Generates a new topic structured as a {@link Map}
    public Map<String, Object> generateNewTopic() {
        Map<String, Object> answer = new HashMap<String, Object>();
        answer.put("TopicId", ran.nextInt());
        answer.put("TopicName", "Camel in Action");
        answer.put("url",  "Camel in Action" );
        answer.put("ModuleId", ran.nextInt());
        answer.put("CreateDate", new Date());
        return answer;

     * Processes the NewTopic
     * @param data  the NewTopic as a {@link Map}
     * @return the transformed NewTopic
    public String processNewTopic(Map<String, Object> data) {
        return "Processed NewTopic id " + data.get("TopicId") + " TopicName " 
		+ data.get("TopicName")
		+ " of " + data.get("ModuleId") + " copies of " + data.get("url");

## notice we use named parameters in the queries, eg :#name.
## A named query parameter must start with :#
## sql that insert new orders
sql.insertNewTopic=INSERT INTO newtopic(TopicId, TopicName, url, ModuleId, CreateDate) VALUES
(:#TopicId, :#TopicName, :#url, :#ModuleId, :#CreateDate)

## sql that select all unprocessed NewTopics
sql.selectNewTopic=select * from newtopic
## sql that update the NewTopic as being processed
sql.markNewTopic=update newtopic set TopicName = 'Apache Camel' where TopicId = :#TopicId

# The logging properties
log4j.rootLogger=INFO, out


# CONSOLE appender not used by default
log4j.appender.out.layout.ConversionPattern=[%30.30t] %-30.30c{1} %-5p %m%n
#log4j.appender.out.layout.ConversionPattern=%d [%-15.15t] %-5p %-30.30c{1} - %m%n

Router in Xml Configuration:(applicationContext-camel.xml)
<beans xmlns="http://www.springframework.org/schema/beans"

	<bean id="dataSource" class="org.apache.commons.dbcp.BasicDataSource"
		<property name="driverClassName" value="com.mysql.jdbc.Driver" />
		<property name="url" value="jdbc:mysql://localhost/javavill_forum" />
		<property name="username" value="root" />
		<property name="password" value="" />

	<!-- configure the Camel SQL component to use the JDBC data source -->
	<bean id="sqlComponent" class="org.apache.camel.component.sql.SqlComponent">
		<property name="dataSource" ref="dataSource" />
	<bean id="topicBean" class="com.javavillage.camel.sql.NewTopicBean" />

	<!-- here is Camel configured with a number of routes -->
	<camelContext xmlns="http://camel.apache.org/schema/spring">

		<!-- use Camel property placeholder loaded from the given file -->
		<propertyPlaceholder id="placeholder" location="classpath:sql.properties" />

		<!-- route that generate new orders and insert them in the database -->
		<route id="generateOrder-route">
			<from uri="timer:foo?period=5s" />
				<method ref="topicBean" method="generateNewTopic" />
			<to uri="sqlComponent:{{sql.insertNewTopic}}" />
			<log message="Inserted new NewTopic ${body[TopicId]}" />

			route that process the NewTopics by picking up new rows from the
			database and when done processing then update the row to mark it as
		<route id="processNewTopic-route">
			<from uri="sqlComponent:{{sql.selectNewTopic}}?
						consumer.onConsume={{sql.markNewTopic}}" />
			<to uri="bean:topicBean?method=processNewTopic" />
			<log message="${body}" />
			<log message="Updated new NewTopic "/>


Maven entries for pom.xml



Below is my application execution
package com.javavillage.camel.proj;

import org.springframework.context.support.AbstractApplicationContext;
import org.springframework.context.support.ClassPathXmlApplicationContext;

 * Hello world!
public class App 
    public static void main( String[] args )
    	AbstractApplicationContext ctx = new ClassPathXmlApplicationContext
    	try {
		} catch (InterruptedException e) {

Apache Camel with SQL example Application Structure

Apache Camel with sql example Application Structure

Execute Apache Camel with SQL Application:

Execution Apache Camel with SQL Application

Insert query will be called based on the timer and update query will be no time limit, same thing can be abserved from screenshot.