Using aggregate in the CXFRS camel route example

In the following post I will demonstrates using camel aggregate .
The example accumulates the massages body into comma separated string .
When the no of aggregation events reaches 3 it output the aggregate string into the log.
The simple aggregator:

package org.apache.camel.example.cxf;
import org.apache.camel.Exchange;
import org.apache.camel.processor.aggregate.AggregationStrategy;
public class myAggregationStrategy  implements AggregationStrategy {
 
    public Exchange aggregate(Exchange oldExchange, Exchange newExchange) {
        
        int theMessageCounter = 0 ; 
        String theMsgOld = "";
        if (oldExchange != null) {           
            theMessageCounter = oldExchange.getIn().getHeader("MessageCounter", Integer.class);
        
            theMsgOld = oldExchange.getIn().getBody(String .class);
        }
        
        theMessageCounter = theMessageCounter + 1;
        
        String thenewMsg =  newExchange.getIn().getBody(String .class);
        
        String theMsgToReturn = theMsgOld;
        
        if ( !"".equals(theMsgOld))
        {
            theMsgToReturn += ",";
        }
        
        theMsgToReturn += thenewMsg;
                
        newExchange.getIn().setBody (theMsgToReturn);
        
        newExchange.getIn().setHeader ( "MessageCounter" , Integer.toString( theMessageCounter) );
        
        return newExchange;
    } 
}

The Route builder :

package org.apache.camel.example.cxf;
import java.util.Map;
import java.util.logging.Level;
import java.util.logging.Logger;
import javax.ws.rs.core.Response;
import org.apache.camel.CamelContext;
import org.apache.camel.Exchange;
import org.apache.camel.LoggingLevel;
import org.apache.camel.Message;
import org.apache.camel.Processor;
import org.apache.camel.builder.RouteBuilder;
import org.apache.camel.processor.interceptor.DefaultTraceFormatter;
import org.apache.camel.processor.interceptor.Tracer;
    public class CamelRoute extends RouteBuilder {
            
    private String uri ="cxfrs:/myData?resourceClasses=org.apache.camel.example.cxf.RsService&bindingStyle=SimpleConsumer";
  
     static Logger myLogger = Logger.getLogger(
                      CamelRoute.class.getName());
    @Override
    public void configure() throws Exception {   
              
        from("seda:tap").aggregate(body(), new myAggregationStrategy()).
            completionPredicate(header("MessageCounter").in(3))
                 .log(LoggingLevel.INFO , "the body ${body}");
        from(uri)
		  .wireTap("seda:tap")
		  .process(new Processor() {
            public void process(Exchange exchange) throws Exception { 
                   Message inMessage = exchange.getIn();                        
                 
                    // The parameter of the invocation is stored in the body of in message
                    String theBody = inMessage.getBody(String.class);
                    theResult += "The Body:" + theBody;
                    myLogger.setLevel (Level.INFO );
                    Response r = Response.ok("This is the result\r\n" +theBody).status(200).build();                                                              
                          
                    exchange.getOut().setBody(r);                 
                    }                     
                });

There is an additional route that is created in order to handle the aggregation.
A copy of the original message is send to the route using the Seda .
The aggregator class aggregate the body of the original messages into comma delimited string and attach a message header with a counter variable that counts the no of execution of the aggregator .
The route checks the counter header and if its equal to 3 then the output of the aggregator is flashed away to the log component .
Note the a new instance of the aggregator state is created after the flashing .

Test rest calls:
/camel-example-cxfrs-tomcat/webservices/myData/RsService/SayHello/Cat
/camel-example-cxfrs-tomcat/webservices/myData/RsService/SayHello/Dog
/camel-example-cxfrs-tomcat/webservices/myData/RsService/SayHello/Pig

And the Result in the log :
2014-04-25 16:32:37,876 [#0 - seda://tap] INFO  route1                         - the body Cat,Dog,Pig

Note:
The is a timeout issue that resets the aggregation component state if  a message is not reached in a time period .

Add Debugging logging and tracing to my Camel cxfrs rest server

In order to debug the camel routes and find why there problem transferring or processing messages there is an option to set tracing on camel .

Adding tracing to the camel-config.Xml file :

<bean id="camelTracer" class="org.apache.camel.processor.interceptor.Tracer">
    <property name="traceExceptions" value="true"/>
    <property name="traceInterceptors" value="true"/>
    <property name="logLevel" value="INFO"/>
    <property name="logName" value="com.mycompany.messages"/>
</bean>
  
<camelContext xmlns="http://camel.apache.org/schema/spring" trace="true">

Add to the log4j configuration file appender in order to make the log more readably

log4j.appender.file=org.apache.log4j.RollingFileAppender
log4j.appender.file.File=D:\\Temp\\loging.txt
log4j.appender.file.MaxFileSize=10MB
log4j.appender.file.MaxBackupIndex=1
log4j.appender.file.layout=org.apache.log4j.PatternLayout
log4j.appender.file.layout.ConversionPattern=%d [%-15.15t] %-5p %-30.30c{1} - %m%n

In the route builder in the configure method we can call to set the logging output formatting:

 @Override
    public void configure() throws Exception {   
        
        Tracer tracer = new Tracer();
        tracer.setTraceOutExchanges(true);
        // we configure the default trace formatter where we can
        // specify which fields we want in the output
        DefaultTraceFormatter formatter = new DefaultTraceFormatter();
        formatter.setShowOutBody(true);
        formatter.setShowOutBodyType(true);
        formatter.setShowException(true);
        formatter.setShowExchangePattern(true);
        formatter.setShowProperties(true);
        formatter.setShowOutHeaders(true);
        // set to use our formatter
        tracer.setFormatter(formatter);
        getContext().addInterceptStrategy(tracer);
.
.
.
.

The logging information is dump to both the tomcat server console log and to the log file that has bean declared in the log4j configuration file :
2014-04-25 11:46:28,576 [io-8080-exec-42] INFO  messages                       - ID-Zvika-PC-2757-1398415530094-0-14 >>> (route2) wireTap(Endpoint[seda://tap]) --> org.apache.camel.example.cxf.CamelRoute$1@1a304efc <<< Pattern:InOut, Headers:{breadcrumbId=ID-Zvika-PC-2757-1398415530094-0-13, connection=Keep-Alive, user-agent=Apache-HttpClient/4.1.1 (java 1.5), CamelCxfRsResponseGenericType=class javax.ws.rs.core.Response, CamelCxfRsOperationResourceInfoStack=[org.apache.cxf.jaxrs.model.MethodInvocationInfo@6b6e21bb], CamelAcceptContentType=*/*, CamelCxfMessage=org.apache.cxf.message.XMLMessage@99357e13, operationName=SayHello, CamelCxfRsResponseClass=class javax.ws.rs.core.Response, Name=EtiGoldfarb, CamelHttpPath=/RsService/SayHello/EtiGoldfarb, accept-encoding=gzip,deflate, CamelHttpMethod=GET, host=localhost:8080, CamelHttpUri=/camel-example-cxfrs-tomcat/webservices/myData/RsService/SayHello/EtiGoldfarb}, BodyType:org.apache.cxf.message.MessageContentsList, Body:EtiGoldfarb
2014-04-25 11:46:28,578 [#0 - seda://tap] INFO  messages                       - ID-Zvika-PC-2757-1398415530094-0-16 >>> (route1) from(seda://tap) --> log[the body ${body}] <<< Pattern:InOnly, Headers:{CamelHttpUri=/camel-example-cxfrs-tomcat/webservices/myData/RsService/SayHello/EtiGoldfarb, CamelHttpPath=/RsService/SayHello/EtiGoldfarb, host=localhost:8080, CamelAcceptContentType=*/*, operationName=SayHello, CamelCxfRsResponseGenericType=class javax.ws.rs.core.Response, breadcrumbId=ID-Zvika-PC-2757-1398415530094-0-13, CamelCxfRsResponseClass=class javax.ws.rs.core.Response, connection=Keep-Alive, accept-encoding=gzip,deflate, CamelCxfRsOperationResourceInfoStack=[org.apache.cxf.jaxrs.model.MethodInvocationInfo@6b6e21bb], CamelHttpMethod=GET, Name=EtiGoldfarb, user-agent=Apache-HttpClient/4.1.1 (java 1.5), CamelCxfMessage=org.apache.cxf.message.XMLMessage@99357e13}, BodyType:org.apache.cxf.message.MessageContentsList, Body:EtiGoldfarb

Example for detailed exception in the log :
org.apache.camel.RuntimeCamelException: org.apache.camel.FailedToCreateRouteException: Failed to create route route1 at: >>> From[seda:tap2] <<< in route: Route(route1)[[From[seda:tap2]] -> []] because of Route route1 has no output processors. You need to add outputs to the route such as to("log:foo").

Camel cxfrs headers

A new method is added to the cxfrs contract in order to show the POST method

package org.apache.camel.example.cxf;
import javax.ws.rs.GET;
import javax.ws.rs.POST;
import javax.ws.rs.Path;
import javax.ws.rs.PathParam;
import javax.ws.rs.core.Response;
@Path("/RsService/")
public interface RsService{
    @GET
    @Path("/SayHello/{Name}")
public Response SayHello (@PathParam("Name") String pName);
    @POST
    @Path("/SayByeBye/")
public Response SayByeBye ();
}

And add logging for the headers in the camel exchange :

package org.apache.camel.example.cxf;
import java.util.Map;
import java.util.logging.Level;
import java.util.logging.Logger;
import javax.ws.rs.core.Response;
import org.apache.camel.Exchange;
import org.apache.camel.Message;
import org.apache.camel.Processor;
import org.apache.camel.builder.RouteBuilder;
public class CamelRoute extends RouteBuilder {
            
    private String uri ="cxfrs:/myData?resourceClasses=org.apache.camel.example.cxf.RsService&bindingStyle=SimpleConsumer";
  
     static Logger myLogger = Logger.getLogger(
                      CamelRoute.class.getName());
    
    @Override
    public void configure() throws Exception {        
          from(uri)          
                 .process(new Processor() {
                      public void process(Exchange exchange) throws Exception { 
                          
                             Message inMessage = exchange.getIn();                        
                                    
                             Map<String, Object> theHeaders = inMessage.getHeaders();
                   
                             String theResult= ""; 
                    
                            for ( String NextHeader : theHeaders.keySet())
                            {
                                theResult += NextHeader + ":" + theHeaders.get(NextHeader) + "\r\n";
                            }
                            
                            String theBody = inMessage.getBody(String.class);
                      
                            theResult += "The Body:" + theBody;
                    
                            myLogger.setLevel (Level.INFO );
                    
                            myLogger.log(Level.INFO , theResult );
                    
                            Response r = Response.ok("This is the result\r\n" +theBody).status(200).build();                                                              
                             exchange.getOut().setBody(r);                 
                    }                     
                });
         
    }
}

And the Results: (The get method )

The logging:

The Post method

The logging

Simple Rest Camel Service

The rest service is build using apache :Camel , Cxf , and tomcat service .

The artifacts
 

The pom file:

<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/maven-v4_0_0.xsd">
    <modelVersion>4.0.0</modelVersion>
    <parent>
        <groupId>org.apache.camel</groupId>
        <artifactId>examples</artifactId>
        <version>2.13.0</version>
    </parent>
    <artifactId>camel-example-cxfrs-tomcat</artifactId>
    <name>Camel :: Example :: CXFRS :: Tomcat</name>
    <description>An example using Camel CXFRS  with Apache Tomcat</description>
    <packaging>war</packaging>
    <build>
        <plugins>
            <plugin>
                <groupId>org.codehaus.mojo</groupId>
                <artifactId>tomcat-maven-plugin</artifactId>
                <version>1.1</version>
                <configuration>
                    <server>TomcatServer</server>
                    <url>${tomcat.url}</url>
                    <path>/${project.build.finalName}</path>
                </configuration>
            </plugin>
            <plugin>
                <groupId>org.codehaus.mojo</groupId>
                <artifactId>exec-maven-plugin</artifactId>
                <configuration>
                    <mainClass>org.apache.camel.example.cxf.CamelRouteClient</mainClass>
                    <includePluginDependencies>false</includePluginDependencies>
                    <systemProperties>
                        <property>
                            <key>java.util.logging.config.file</key>
                            <value>logging.properties</value>
                        </property>
                    </systemProperties>
                </configuration>
            </plugin>
        </plugins>
        <!-- Name of the generated WAR file -->
        <finalName>camel-example-cxfrs-tomcat</finalName>
    </build>
    <dependencies>
        <!-- camel -->
        <dependency>
            <groupId>org.apache.camel</groupId>
            <artifactId>camel-core</artifactId>
        </dependency>
        <dependency>
            <groupId>org.apache.camel</groupId>
            <artifactId>camel-spring</artifactId>
        </dependency>
        <dependency>
            <groupId>org.apache.camel</groupId>
            <artifactId>camel-cxf</artifactId>
        </dependency>
        <dependency>
  <groupId>org.apache.camel</groupId>
  <artifactId>camel-gson</artifactId>
 
</dependency>
        <dependency>
            <groupId>org.apache.camel</groupId>
            <artifactId>camel-servlet</artifactId>
        </dependency>
        <dependency>
            <groupId>org.springframework</groupId>
            <artifactId>spring-web</artifactId>
        </dependency>
        
        <!-- cxf -->
        <dependency>
            <groupId>org.apache.cxf</groupId>
            <artifactId>cxf-rt-frontend-jaxws</artifactId>
            <version>${cxf-version}</version>
        </dependency>
        <dependency>
            <groupId>org.apache.cxf</groupId>
            <artifactId>cxf-rt-transports-http</artifactId>
            <version>${cxf-version}</version>
        </dependency>
        <!-- logging -->
        <dependency>
            <groupId>log4j</groupId>
            <artifactId>log4j</artifactId>
        </dependency>
        <dependency>
            <groupId>org.slf4j</groupId>
            <artifactId>slf4j-log4j12</artifactId>
        </dependency>
    </dependencies>
    <profiles>
        <profile>
            <id>Tomcat7</id>
            <activation>
                <activeByDefault>true</activeByDefault>
            </activation>
            <properties>
                <tomcat.url>http://localhost:8080/manager/text</tomcat.url>
            </properties>
        </profile>
        <profile>
            <id>Tomcat6</id>
            <properties>
                <tomcat.url>http://localhost:8080/manager</tomcat.url>
            </properties>
        </profile>
    </profiles>
</project>

The web.xml

<web-app version="2.4" xmlns="http://java.sun.com/xml/ns/j2ee"
		xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
		xsi:schemaLocation="http://java.sun.com/xml/ns/j2ee http://java.sun.com/xml/ns/j2ee/web-app_2_4.xsd">
	<display-name>My Web Application</display-name>
	<!-- location of spring xml files -->
	<context-param>
		<param-name>contextConfigLocation</param-name>
        <param-value>classpath:camel-config.xml</param-value>
	</context-param>
	<!-- the listener that kick-starts Spring -->
	<listener>
		<listener-class>org.springframework.web.context.ContextLoaderListener</listener-class>
	</listener>
	<!-- CXF servlet -->
	<servlet>
		<servlet-name>CXFServlet</servlet-name>
		<servlet-class>org.apache.cxf.transport.servlet.CXFServlet</servlet-class>
		<load-on-startup>1</load-on-startup>
		<!-- If you want to leverage the Servlet3's async feature in Tomcat, 
		 please enable this feature 
		<async-supported>true</async-supported>
		-->
	</servlet>
	<!-- all our webservices are mapped under this URI pattern -->
	<servlet-mapping>
		<servlet-name>CXFServlet</servlet-name>
		<url-pattern>/webservices/*</url-pattern>
	</servlet-mapping>
</web-app>

The web xml file declares a route for the tomcat server :
When the tomcat receives the following url pattern:/webservices/* send the message to the cxf servlet.(org.apache.cxf.transport.servlet.CXFServlet)

Although the web config  kicks off the camel configuration procedure .

The Camel config.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"
       xmlns:cxf="http://camel.apache.org/schema/cxf"
       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
       http://camel.apache.org/schema/cxf http://camel.apache.org/schema/cxf/camel-cxf.xsd">
  <!-- We still need it for loading the CXFServlet --> 
  <import resource="classpath:META-INF/cxf/cxf.xml"/>
  
  <bean id="myRoutes" class="org.apache.camel.example.cxf.CamelRoute"/>
    
  <camelContext xmlns="http://camel.apache.org/schema/spring">
  
   <routeBuilder ref="myRoutes"/> 
   
  </camelContext>
</beans>

Declare a route builder bean .

The camel route

package org.apache.camel.example.cxf;
import javax.ws.rs.core.Response;
import org.apache.camel.Exchange;
import org.apache.camel.Processor;
import org.apache.camel.builder.RouteBuilder;
public class CamelRoute extends RouteBuilder {
            
    private String uri ="cxfrs:/myData?resourceClasses=org.apache.camel.example.cxf.RsService&bindingStyle=SimpleConsumer";
  
    @Override
    public void configure() throws Exception {        
          from(uri)          
                 .process(new Processor() {
                      public void process(Exchange exchange) throws Exception { 
                                Response r = Response.ok("This is the result").status(200).build();                                                              
                                exchange.getOut().setBody(r);                 
                    }                     
                });
         
    }
}

The resource (the interface declaration)

package org.apache.camel.example.cxf;
import javax.ws.rs.GET;
import javax.ws.rs.Path;
import javax.ws.rs.core.Response;
@Path("/RsService/")
public interface RsService{
    @GET
    @Path("/SayHello/")
public Response SayHello ();
}

The project

running

Execute the project using the maven Tomcat:Redeploy goal

soap ui

The resource :/camel-example-cxfrs-tomcat/webservices/myData/RsService/SayHello