Clone and Aggregate Integration Pattern with Ballerina

Clone and Aggregate integration pattern is a very common pattern in the world of integration. Basically, it is another form of service orchestration in which the caller payload is cloned into one or more similar payloads, transformed and sent to different back-ends. Being a cloud native language, Ballerina can help you implement such pattern within minutes. This article is a short guide on how to implement Clone and Aggregate integration pattern using Ballerina 0.990.0.

Ballerina 0.990.0 was a major release that introduced a significant number of new features and changes to the language. Do check out the release notes for Ballerina 0.990.0 for detailed descriptions on the changes.

Consider the below simple use-case. Client sends a JSON request to Ballerina service. Ballerina service clones the client payload and transforms before sending it to the relevant back-ends in parallel. Once the back-end responses are received those responses are aggregated and sent back to the client.

This article contains how similar scenario is implemented in WSO2 ESB.

Let’s start with the Service

Following is a basic RESTful service with one resource, which invokes two back-ends. Name of the service is clone whereas the name of resource is aggregate. Hence, the resource is uniquely identified using the below URL.

 http://localhost:9090/clone/aggregate

At the service level two outbound client endpoints are created to invoke the aforementioned two different back-ends. Resource is annotated with data binding configuration so that the inbound payload is directly mapped into expected payload type.

Clone and Aggregate Payload

Now, let’s make the resource more interesting by implementing the cloneAndAggregate function, that is there in the resource. First line of the function clones the payload into another variable name callerPayload.

clone() is the equivalent of clone mediator in WSO2 ESB.

Now comes the interesting part, fork construct can have any number of workers, which can execute in parallel. In the above scenario, two workers are created within the fork, each worker transforms the cloned payload and invokes the relevant back-end using invokeEndpoint function. Till the responses are collected from the workers, execution is halted at wait construct. Once the responses are collected, each response is mapped into result record. Using the result record collected JSON payloads are mapped into a JSON array and returned to the resource.

wait is the equivalent of aggregate mediator of WSO2 ESB.

Invoking the Back-end

Lastly, let’s complete the program by implementing the invokeEndpoint function of cloneAndAggregate function. Following is the implementation,

invokeEndpoint function makes a post request to the given back-end and retrieves the response. Inbound response contains JSON payload which is returned back to the caller. In case of an error, JSON payload with error message is returned back to the caller.

Trying out the Service

That is it. Lastly, to try out the service put the aforementioned bits into one Ballerina file and start the program. Below request can be used to invoke the Ballerina program.

curl -v localhost:9090/clone/aggregate -H “Content-Type: application/json” -d “{\”key1\”: \”value1\”}”

In case you need a back-end, use the below command to pull the echo back-end.

docker pull shafreen/netty_echo_server:latest

Then start the back-ends with the relevant port numbers as below,

docker run -d -p 8601:8688 netty_echo_server
docker run -d -p 8602:8688 netty_echo_server

Conclusion

As you can see, using Ballerina, one can easily implement Clone and Aggregate integration pattern. Even though above example only have two back-ends, it can be altered to have many back-ends. Therefore, this example can be used as the base and then altered accordingly to fit your requirement. For more integration scenarios please check Ballerina by Guide.