I've demonstrated how to do event notification with Apache Camel before. But the classes were hard-coupled to Camel URI endpoints.
This time, I have these goals in mind:
- Decoupled. The event source and listener(s) must be as decoupled as possible, but in a good way that it's not too complex.
- As POJO as possible. Event sources, event classes, and event listeners need not import the event framework classes/interfaces. Annotations may be used.
The Big Picture
I will illustrate this technique using a sensor and display simulation application. The sensor (supposedly) detects a text, and when the text is changed, the sensor will notify its listeners. The display(s), as listeners, responds to such events and updates its own display.Sensors and Displays do not need to know each other, beyond mutually agreed Listener interfaces.
I've implented a runnable proof-of-concept application and as such this technique can practically be used.
You can download/explore the example from: http://github.com/ceefour/eventfx.camel
Listener Interfaces
The Listener interfaces provide event sources and event. I'm using generic listener interfaces to avoid creating too many interfaces just for different events. It's also possible to use generic Event payload classes/interfaces if we want to go that far.Let's define a generic listener interface:
package com.soluvas.samples.eventfx.camel;
import java.util.EventListener;
import org.apache.camel.InOnly;
public interface AsyncListener<E> extends EventListener {
@InOnly
void notify(E event);
}This defines our asynchronous listener. That is, a fire-and-forget type of event. The event source does not need a response in this case.Let's define another generic interface for synchronous events, that provides a response:
package com.soluvas.samples.eventfx.camel;
import java.util.EventListener;
import org.apache.camel.InOut;
public interface SyncListener<E, R> extends EventListener {
@InOut
R update(E event);
}
I'm going to use a simple Event payload class, which contains a single String text property. If there are many events like this, we can potentially use a generic event class.package com.soluvas.samples.eventfx.camel;
public class SensorEvent {
private String text;
public SensorEvent(String text) {
super();
this.text = text;
}
public String getText() {
return text;
}
public void setText(String text) {
this.text = text;
}
}
The Event Source: Sensor
Let's create the code for Sensor. Input for this sensor is manually set for the simulation.The sensor can be hooked to a SyncListener
package com.soluvas.samples.eventfx.camel;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
public class Sensor {
private static Log log = LogFactory.getLog(Sensor.class);
private SyncListener<SensorEvent, String> sensorSyncListener;
private AsyncListener<SensorEvent> sensorAsyncListener;
private void fireSensor(SensorEvent event) {
if (sensorSyncListener != null) {
String result = sensorSyncListener.update(event);
log.info("Response: " + result);
}
if (sensorAsyncListener != null) {
sensorAsyncListener.notify(event);
}
}
public void updateText(String text) {
log.info("updateText: " + text);
fireSensor(new SensorEvent(text));
}
public SyncListener<SensorEvent, String> getSensorSyncListener() {
return sensorSyncListener;
}
public void setSensorSyncListener(SyncListener<SensorEvent, String> sensorSyncListener) {
this.sensorSyncListener = sensorSyncListener;
}
public AsyncListener<SensorEvent> getSensorAsyncListener() {
return sensorAsyncListener;
}
public void setSensorAsyncListener(
AsyncListener<SensorEvent> sensorAsyncListener) {
this.sensorAsyncListener = sensorAsyncListener;
}
}
The Event Listener: Display
The Display implements both AsyncListener<SensorEvent> and SyncListener<SensorEvent, String> interfaces, so it can be used for both purposes.package com.soluvas.samples.eventfx.camel;
import javax.annotation.PostConstruct;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
public class Display implements SyncListener<SensorEvent, String>, AsyncListener<SensorEvent> {
private static Log log = LogFactory.getLog(Display.class);
private String name;
@PostConstruct
public void initialize() {
log.info("Display " + getName() + " created.");
}
@Override
public String update(SensorEvent event) {
log.info("[" + name + "] is updated: '" + event.getText() + "'");
return getName() + " received " + event.getText();
}
@Override
public void notify(SensorEvent event) {
log.info("[" + name + "] is notified: '" + event.getText() + "'");
}
public String getName() {
return name;
}
public void setName(String name) {
this.name = name;
}
}
Implementation of the display is simply to log the SensorEvent's text with INFO level.Simulating Sensor Changes
Since the Sensor must be manually updated, I created a simulator class that periodically updates the sensor with current time.package com.soluvas.samples.eventfx.camel;
import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
@Component
public class SensorSimulator {
private static Log log = LogFactory.getLog(SensorSimulator.class);
@Autowired
private Sensor sensor;
private ScheduledExecutorService scheduledExecutor = Executors.newSingleThreadScheduledExecutor();
@PostConstruct
public void initialize() {
scheduledExecutor.scheduleAtFixedRate(new Runnable() {
public void run() {
sensor.updateText("Something happens at " + SimpleDateFormat.getTimeInstance().format(new Date()));
}
}, 2000, 2000, TimeUnit.MILLISECONDS);
log.info("Sensor simulator initialized.");
}
@PreDestroy
public void destroy() throws InterruptedException {
log.info("Destroying Sensor simulator...");
scheduledExecutor.shutdown();
scheduledExecutor.awaitTermination(2000, TimeUnit.MILLISECONDS);
}
}
The Simulator is an autonomous Spring component bean. It will get autowired to a Sensor bean (object) and do its job automatically until it is destroyed (in our case, until the application is terminated.)Wiring The Application
To wire the beans that make up the application, I use Spring Framework 2.5.6 and Apache Camel 2.1.0 together.I put the following inside META-INF/spring/camel-context.xml :
<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns:context="http://www.springframework.org/schema/context"
xmlns:camel="http://camel.apache.org/schema/spring"
xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-2.5.xsd
http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context-2.5.xsd
http://camel.apache.org/schema/spring http://camel.apache.org/schema/spring/camel-spring.xsd">
<context:component-scan base-package="com.soluvas.samples.eventfx.camel" />
<bean class="com.soluvas.samples.eventfx.camel.Sensor">
<property name="sensorSyncListener">
<bean class="org.apache.camel.spring.remoting.CamelProxyFactoryBean">
<property name="serviceUrl" value="seda:Sensor.SensorEventSync"/>
<property name="serviceInterface" value="com.soluvas.samples.eventfx.camel.SyncListener"/>
</bean>
</property>
<property name="sensorAsyncListener">
<bean class="org.apache.camel.spring.remoting.CamelProxyFactoryBean">
<property name="serviceUrl" value="seda:Sensor.SensorEventAsync"/>
<property name="serviceInterface" value="com.soluvas.samples.eventfx.camel.AsyncListener"/>
</bean>
</property>
</bean>
<bean id="display1" class="com.soluvas.samples.eventfx.camel.Display">
<property name="name" value="Sony(sync)" />
</bean>
<bean id="display2" class="com.soluvas.samples.eventfx.camel.Display">
<property name="name" value="Samsung(async)" />
</bean>
<camelContext xmlns="http://camel.apache.org/schema/spring">
<packageScan>
<package>com.soluvas.samples.eventfx.camel</package>
</packageScan>
</camelContext>
</beans>
The above Spring configuration file sets up:- autodetection of stereotype-annotated (@Component) components
- a Sensor bean that gets wired to Camel proxy listeners.
The sync events go to seda:Sensor.SensorEventSync
The async events go to seda:Sensor.SensorEventAsync - two Display beans, named Sony and Samsung (I don't endorse any of them)
- Camel context bean, which will scan com.soluvas.samples.eventfx.camel package for route builders
- Spring context bootstrapping
- RouteBuilder configuration
The code for MainRouteBuilder.java is as follows:
package com.soluvas.samples.eventfx.camel;
import org.apache.camel.builder.RouteBuilder;
import org.springframework.context.support.ClassPathXmlApplicationContext;
public class MainRouteBuilder extends RouteBuilder {
/**
* A main() so we can easily run these routing rules in our IDE
*/
public static void main(String... args) {
new ClassPathXmlApplicationContext("META-INF/spring/*.xml");
}
/**
* Lets configure the Camel routing rules using Java code...
*/
public void configure() {
from("seda:Sensor.SensorEventSync").beanRef("display1");
from("seda:Sensor.SensorEventAsync").delay(1200).beanRef("display2");
}
}
The RouteBuilder implementation (configure()) above configures:- routes seda:SensorEventSync events to display1 (Sony).
- delays seda:SensorEventAsync events by 1,200 milliseconds (to simulate asynchronous behavior) and routes them to display2 (Samsung).
Camel-based Event Notification Framework in Action
It's hard to visualize this application in writing (you have to see it running to notice the event timings). But this is the application output:[ main] CamelNamespaceHandler INFO camel-osgi.jar not detected in classpath [ main] SensorSimulator INFO Sensor simulator initialized. [ main] Display INFO Display Sony(sync) created. [ main] Display INFO Display Samsung(async) created. [ main] DefaultCamelContext INFO Apache Camel 2.1.0 (CamelContext:camelContext) is starting [ main] DefaultCamelContext INFO JMX enabled. Using DefaultManagedLifecycleStrategy. [ main] DefaultCamelContext INFO Apache Camel 2.1.0 (CamelContext:camelContext) started [ pool-1-thread-1] Sensor INFO updateText: Something happens at 6:17:04 PM [ seda://Sensor.SensorEventSync] Display INFO [Sony(sync)] is updated: 'Something happens at 6:17:04 PM' [ pool-1-thread-1] Sensor INFO Response: Sony(sync) received Something happens at 6:17:04 PM [seda://Sensor.SensorEventAsync] Display INFO [Samsung(async)] is notified: 'Something happens at 6:17:04 PM' [ pool-1-thread-1] Sensor INFO updateText: Something happens at 6:17:06 PM [ seda://Sensor.SensorEventSync] Display INFO [Sony(sync)] is updated: 'Something happens at 6:17:06 PM' [ pool-1-thread-1] Sensor INFO Response: Sony(sync) received Something happens at 6:17:06 PMWe have successfully wired the Simulator, Sensor, and Displays, through Apache Camel.
The Sensor does not know about the Displays. The Displays do not know whether they're being wired with Sensor, only that they can act as SensorEvent listeners (both async and sync).
Neither Sensor nor Display are aware that they're being connected via Apache Camel.
During firing of asynchronous event, Sensor expects the method call to return immediately. So even that we have delayed the async event for 1,200 milliseconds, Sensor could have proceeded with other processing during that time.
Demonstrating Camel's Flexibility: Multicasting
I know you're smart enough to realize that Camel was only used as infrastructure and are curious, "but it can be done without Camel!" and you'll be right... with some gotchas.You can leave out Camel and using Spring, wire directly the event source with its listeners. With synchronous events, it's doable and may even be exactly how you want to do it ("Keep It Simple, Stupid" principle).
With asynchronous events, on the other hand, it's going to take you more work. Of course, you can code listeners to be asynchronously, but I think it's better to leave that job to Camel's configuration (routes).
For example, if I want to do configure the sensor so all events are asynchronous and "multicasted" to both display1 and display2, I simply change the Camel route configuration to:
public void configure() {
ProcessorDefinition def = from("seda:Sensor.SensorEventAsync").multicast();
def.beanRef("display1");
def.delay(700).beanRef("display2");
}
And that's it. No need to change the implementation of Display (to pass the call to next in chain) nor the Sensor to handle multiple asynchronous listeners (traditionally, you'd need to use a List of listeners).Conclusion: Implementing Event Notification with Apache Camel
Using Apache Camel proxy beans instead of plain Java (or plain Spring) method calls for event notification allows you to (but not limited to the following):- Still keep your POJO interfaces and method calls.
- Switch between asynchronous and synchronous calls as you will.
- Route the events to practically anywhere you want.
- Multicast events and other routing techniques.
- Transport events through network, via HTTP/email/XMPP/JMS or other protocols supported by Camel.
- Decouple not only between event sources and event listeners, but also event routing.
Learning More on Enterprise Integration
Apache Camel is all about message routing, which is part of enterprise integration. If you want to learn more about enterprise integration, I heartily recommend the book Enterprise Integration Patterns: Designing, Building, and Deploying Messaging Solutions
.
Great walkthrough Hendy, thanks. I'm just getting started with Camel, and the flexibility to do things like event-based processing is what attracted me to the framework.
ReplyDelete