- Home
- Integration Frameworks
- Apache Camel
- 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..
NewTopicBean.java
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");
}
}
sql.properties
## 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
log4j.properties
#
# The logging properties
#
log4j.rootLogger=INFO, out
#log4j.logger.org.apache.camel.component.sql=DEBUG
# CONSOLE appender not used by default
log4j.appender.out=org.apache.log4j.ConsoleAppender
log4j.appender.out.layout=org.apache.log4j.PatternLayout
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"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns:camel="http://camel.apache.org/schema/spring"
xsi:schemaLocation="http://www.springframework.org/schema/beans
http://www.springframework.org/schema/beans/spring-beans.xsd
http://camel.apache.org/schema/spring
http://camel.apache.org/schema/spring/camel-spring.xsd">
<bean id="dataSource" class="org.apache.commons.dbcp.BasicDataSource"
destroy-method="close">
<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="" />
</bean>
<!-- 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>
<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" />
<transform>
<method ref="topicBean" method="generateNewTopic" />
</transform>
<to uri="sqlComponent:{{sql.insertNewTopic}}" />
<log message="Inserted new NewTopic ${body[TopicId]}" />
</route>
<!--
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
processed
-->
<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 "/>
</route>
</camelContext>
</beans>
Maven entries for pom.xml
<dependency>
<groupId>org.apache.camel</groupId>
<artifactId>camel-core</artifactId>
<version>2.13.0</version>
</dependency>
<dependency>
<groupId>org.apache.camel</groupId>
<artifactId>camel-spring</artifactId>
<version>2.13.0</version>
</dependency>
<dependency>
<groupId>org.apache.camel</groupId>
<artifactId>camel-jdbc</artifactId>
<version>2.13.0</version>
</dependency>
<dependency>
<groupId>org.apache.camel</groupId>
<artifactId>camel-sql</artifactId>
<version>2.13.0</version>
</dependency>
<dependency>
<groupId>commons-dbcp</groupId>
<artifactId>commons-dbcp</artifactId>
<version>1.4</version>
</dependency>
<dependency>
<groupId>mysql</groupId>
<artifactId>mysql-connector-java</artifactId>
<version>5.1.6</version>
</dependency>
<dependency>
<groupId>org.slf4j</groupId>
<artifactId>slf4j-api</artifactId>
<version>1.7.5</version>
</dependency>
<dependency>
<groupId>org.slf4j</groupId>
<artifactId>slf4j-log4j12</artifactId>
<version>1.7.5</version>
</dependency>
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
("applicationContext-camel.xml");
ctx.start();
try {
Thread.sleep(5000);
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println("Entered>>>>>");
ctx.stop();
}
}
Apache Camel with SQL example Application Structure
Execute 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.