If you want to monitor, debug, troubleshoot messages that are flowing through the route without the bother of permanently consuming the message off the channel, then you need to use a wire tap.
The wire tap acts as a recipient list that consumes messages off the input channel and publishes it to both output channels.
The first would be to the actual destination which acts as the primary channel and second one to the wire tap destination which acts as the secondary channel.
Before we start with the example, let’s look into the setup details.
This example uses the following frameworks:
- Maven 3.2.3
- Apache Camel 2.15.1
- Spring 4.1.5.RELEASE
- Eclipse as the IDE, version Luna 4.4.1.
Dependencies
We are just relying camel’s core components and the logger component so our pom.xml consists of:
camel-coreslf4j-apislf4j-log4j12
pom.xml:
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd"> <modelVersion>4.0.0</modelVersion> <groupId>com.javarticles.camel</groupId> <artifactId>camelHelloWorld</artifactId> <version>0.0.1-SNAPSHOT</version> <dependencies> <dependency> <groupId>org.apache.camel</groupId> <artifactId>camel-core</artifactId> <version>2.15.1</version> </dependency> <dependency> <groupId>org.slf4j</groupId> <artifactId>slf4j-api</artifactId> <version>1.7.12</version> </dependency> <dependency> <groupId>org.slf4j</groupId> <artifactId>slf4j-log4j12</artifactId> <version>1.7.12</version> </dependency> </dependencies> </project>
Simple wire tap Example
The wire tap receives the message, makes a copy of the message and sends it to a wire tap destination. The original exchange continues on through the route to reach the actual destination. Camel doesn’t wait for a response from the wire tap because the wire tap sets the message exchange pattern (MEP) to InOnly.
You need to use wireTap statement, specify the endpoint URI of where to send a copy of the message. The Wire Tap processor, processes it on a separate thread managed by the Camel routing engine.
In our example, we send a message ‘One’ to direct:start to initiate the route. A copy of the message will be sent to wireTap destination direct:tap. The original message message continues in the main route to bean for further processing. MyBean.addTwo adds ‘Two’ string to ‘One’. In the wiretap route which happens in a separate thread, the message is sent to MyBean.addThree to add ‘Three’ to ‘One’.
CamelWiretapExample:
package com.javarticles.camel;
import org.apache.camel.CamelContext;
import org.apache.camel.ProducerTemplate;
import org.apache.camel.builder.RouteBuilder;
import org.apache.camel.impl.DefaultCamelContext;
import org.apache.camel.util.jndi.JndiContext;
public class CamelWiretapExample {
public static final void main(String[] args) throws Exception {
JndiContext jndiContext = new JndiContext();
jndiContext.bind("myBean", new MyBean());
CamelContext camelContext = new DefaultCamelContext(jndiContext);
try {
camelContext.addRoutes(new RouteBuilder() {
public void configure() {
from("direct:start")
.log("Main route: Send '${body}' to tap router")
.wireTap("direct:tap")
.log("Main route: Add 'two' to '${body}'")
.bean(MyBean.class, "addTwo")
.log("Main route: Output '${body}'");
from("direct:tap")
.log("Tap Wire route: received '${body}'")
.log("Tap Wire route: Add 'three' to '${body}'")
.bean(MyBean.class, "addThree")
.log("Tap Wire route: Output '${body}'");
}
});
ProducerTemplate template = camelContext.createProducerTemplate();
camelContext.start();
template.sendBody("direct:start", "One");
} finally {
camelContext.stop();
}
}
}MyBean:
package com.javarticles.camel;
import java.util.ArrayList;
import java.util.List;
public class MyBean {
public String addTwo(String body) {
return body + " and two";
}
public String addThree(String body) {
return body + " and three";
}
}The main route final output is ‘One and two’. The wire tap destination output is ‘One and three’.
Output:
12:19| INFO | MarkerIgnoringBase.java 95 | Main route: Send 'One' to tap router 12:19| INFO | MarkerIgnoringBase.java 95 | Main route: Add 'two' to 'One' 12:19| INFO | MarkerIgnoringBase.java 95 | Tap Wire route: received 'One' 12:19| INFO | MarkerIgnoringBase.java 95 | Tap Wire route: Add 'three' to 'One' 12:19| INFO | MarkerIgnoringBase.java 95 | Tap Wire route: Output 'One and three' 12:19| INFO | MarkerIgnoringBase.java 95 | Main route: Output 'One and two'
Shallow copy of message to Wire Tap
The Wire Tap processor, by default, makes a shallow copy of the Camel Exchange instance. The copy of the exchange is sent to the endpoint specified in the wireTap statement. The body of the wire tapped message contains the same object as that in the original message which means any change to the internal state of that object during the wire tap route may also end up changing the main message’s body.
In the below example, instead of sending string ‘One’, we wrap it in a MyPayload object and then send it to the direct:start to initiate the route. The main route appends ‘two’ to the payload’s value, likewise, the wire tap route, appends ‘three’.
MyBean:
package com.javarticles.camel;
import java.util.ArrayList;
import java.util.List;
public class MyBean {
public String addTwo(String body) {
return body + " and two";
}
public String addThree(String body) {
return body + " and three";
}
public MyPayload addTwo(MyPayload body) {
body.setValue(body.getValue() + " and two");
return body;
}
public MyPayload addThree(MyPayload body) {
body.setValue(body.getValue() + " and three");
return body;
}
}MyPayload acts as a wrapper object holding the string value.
MyPayload:
package com.javarticles.camel;
public class MyPayload {
private String value;
public MyPayload(String value) {
this.value = value;
}
public String getValue() {
return value;
}
public void setValue(String value) {
this.value = value;
}
public String toString() {
return value;
}
}Event though the message is copied to wire tap destination, the object it holds is same as the main route’s one. Since wire tap routing is happening concurrently, there is a possibility of it changing the main route’s message.
CamelWiretapShallowCopyExample:
package com.javarticles.camel;
import org.apache.camel.CamelContext;
import org.apache.camel.ProducerTemplate;
import org.apache.camel.builder.RouteBuilder;
import org.apache.camel.impl.DefaultCamelContext;
import org.apache.camel.util.jndi.JndiContext;
public class CamelWiretapShallowCopyExample {
public static final void main(String[] args) throws Exception {
JndiContext jndiContext = new JndiContext();
jndiContext.bind("myBean", new MyBean());
CamelContext camelContext = new DefaultCamelContext(jndiContext);
try {
camelContext.addRoutes(new RouteBuilder() {
public void configure() {
from("direct:start")
.log("Main route: Send '${body}' to tap router")
.wireTap("direct:tap")
.log("Main route: Add 'two' to '${body}'")
.bean(MyBean.class, "addTwo")
.log("Main route: Output '${body}'");
from("direct:tap")
.log("Tap Wire route: received '${body}'")
.log("Tap Wire route: Add 'three' to '${body}'")
.bean(MyBean.class, "addThree")
.log("Tap Wire route: Output '${body}'");
}
});
ProducerTemplate template = camelContext.createProducerTemplate();
camelContext.start();
MyPayload payload = new MyPayload("One");
template.sendBody("direct:start", payload);
System.out.println("Final payload: " + payload.getValue());
} finally {
camelContext.stop();
}
}
}The final payload is corrupted, it is ‘One and three’ instead of ‘One and two’. In our next section, we will deep copy the object before passing it to the wire tap destination.
Output:
15:25| INFO | MarkerIgnoringBase.java 95 | Main route: Send 'One' to tap router 15:25| INFO | MarkerIgnoringBase.java 95 | Main route: Add 'two' to 'One' 15:25| INFO | MarkerIgnoringBase.java 95 | Tap Wire route: received 'One' 15:25| INFO | MarkerIgnoringBase.java 95 | Tap Wire route: Add 'three' to 'One' 15:25| INFO | MarkerIgnoringBase.java 95 | Tap Wire route: Output 'One and three' 15:25| INFO | MarkerIgnoringBase.java 95 | Main route: Output 'One and three' Final payload: One and three 15:25| INFO | DefaultCamelContext.java 2660 | Apache Camel 2.15.1 (CamelCont
Deep Copy of message to Wire Tap
Wire Tap EIP provides us with a mechanism to perform a “deep” copy of the message.
Let’s first add a deep cloning method to MyPayload.
MyPayload:
package com.javarticles.camel;
public class MyPayload {
private String value;
public MyPayload(String value) {
this.value = value;
}
public String getValue() {
return value;
}
public void setValue(String value) {
this.value = value;
}
public String toString() {
return value;
}
public MyPayload deepClone() {
MyPayload myPayload = new MyPayload(value);
return myPayload;
}
}Next, implement a custom Processor to deep clone the MyPayload object.
MyPayloadClonePrepare:
package com.javarticles.camel;
import org.apache.camel.Exchange;
import org.apache.camel.Processor;
public class MyPayloadClonePrepare implements Processor {
public void process(Exchange exchange) throws Exception {
MyPayload myPayload = exchange.getIn().getBody(MyPayload.class);
exchange.getIn().setBody(myPayload.deepClone());
}
}This needs to be be called using onPrepare statement right after wireTap.
CamelWiretapOnPrepareExample:
package com.javarticles.camel;
import org.apache.camel.CamelContext;
import org.apache.camel.ProducerTemplate;
import org.apache.camel.builder.RouteBuilder;
import org.apache.camel.impl.DefaultCamelContext;
import org.apache.camel.util.jndi.JndiContext;
public class CamelWiretapOnPrepareExample {
public static final void main(String[] args) throws Exception {
JndiContext jndiContext = new JndiContext();
jndiContext.bind("myBean", new MyBean());
CamelContext camelContext = new DefaultCamelContext(jndiContext);
try {
camelContext.addRoutes(new RouteBuilder() {
public void configure() {
from("direct:start")
.log("Send '${body}' to tap router")
.wireTap("direct:tap")
.onPrepare(new MyPayloadClonePrepare())
.end()
.delay(1000)
.log("Output of main '${body}'");
from("direct:tap")
.log("Tap router received '${body}'")
.bean(MyBean.class, "addThree")
.log("Output of tap '${body}'");
}
});
ProducerTemplate template = camelContext.createProducerTemplate();
camelContext.start();
MyPayload payload = new MyPayload("One");
template.sendBody("direct:start", payload);
System.out.println("Final payload: " + payload.getValue());
} finally {
camelContext.stop();
}
}
}Now the output of main route is not affected by the wire tap’s route. It correctly shows up as ‘One and two’.
Output:
18:46| INFO | MarkerIgnoringBase.java 95 | Send 'One' to tap router 18:46| INFO | MarkerIgnoringBase.java 95 | Tap router received 'One' 18:46| INFO | MarkerIgnoringBase.java 95 | Output of tap 'One and three' 18:46| INFO | MarkerIgnoringBase.java 95 | Output of main 'One' Final payload: One
Download the source code
This was an example about Apache Camel Wire Tap. You can download the source code here: camelWireTapExample.zip
| Reference: | Apache Camel Wire Tap Examples from our JCG partner Ram Mokkapaty at the Java Articles blog. |
