Building Camel-CXF REST Service in OSGi for Karaf - Multicasting and Aggregation
Join the DZone community and get the full member experience.Join For Free
Please check out my other post on building plain CXF services (without Camel) in OSGi on Karaf.
This is a basic tutorial on how to
- create a CXF REST service
- multicast (and parallelize) the incoming request using Camel
- source data from two different services
- aggregate the response and
- finally return the consolidated result as JSON to the the end user.
You could download the entire codebase from github.
What this application does, in simple terms
The result expected from this service is a hardcoded response which looks like
As you could see from the image, the top portion of the response is sourced from a service called
NameEmailService and the second portion of the response is sourced from a service called
AgePhoneService. The calls to enrich both the data are done concurrently and the consolidated result entity -
ConsolidatedSearchResult is populated.
The Project structure looks like this :
There are two baby steps for Step 1.
Step 1.a - Create a CXF REST Service
As you might have guessed, there’s nothing complicated in this step. Just an interface and an implementation.
Note that the method implementation
sourceResultsFromTwoSources returns a null. The truth is that this method doesn’t even get called when making a REST call. Camel intercepts all requests to the URL and routes it to various endpoints (calls two methods -
getAgePhoneResult(), in our case).
Step 1.b - Create the Service Implementation
Kiddish implementations of the NameEmailService and the AgePhoneService are below :
Step 2, 3, 4 & 5
Well, I lied when I said 2,3,4 and 5 were 4 steps. They are all done as a single step using Camel routing and its Enterprise Integration Pattern implementations.
Our Routing explained
Simply put, what our routerbuilder does is that it
from ("cxfrs://bean://rsServer") Intercepts all requests to a JAX-RS server endpoint defined in the
.multicast() forwards the original request untouched to
1. `getNameEmailResult` & 2. `getAgePhoneResult` methods in `RestServiceImpl`
.parallelProcessing() places concurrent calls to the methods.
.aggregationStrategy(new ResultAggregator()) specifies how the results from various multicasted sources should be, well, aggregated.
Our aggregator looks like :
Our Aggregator explained
aggregate method in our ResultAggregator is a little crude but does the job.
aggregatemethod gets called for all multicasted endpoints whenever they finish.
So, the first time, the oldExchange will be null. We take that as an opportunity to construct the final consolidated result entity that we wanted to respond to the user.
We check whether the newExchange that comes in is the result of a call to the NameEmailService or AgePhoneService and populate the consolidated entity accordingly.
Finally, we return the consolidated entity - the returning does two jobs.
The consolidated entity comes in as oldExchange for the next call to the
aggregatemethod. (more like chaining - the last returned object from the entity is the one which comes in as the incoming exchange for the next call)
Gets returned back to the user if it is the last call of
aggregate(all multicast endpoints calls are complete).
Published at DZone with permission of Arun Manivannan, DZone MVB. See the original article here.
Opinions expressed by DZone contributors are their own.