Using Clone and Aggregate Mediators — WSO2 ESB / WSO2 MI

The Aggregate mediator implements the Aggregator enterprise integration pattern. It combines (aggregates) the response messages of messages that were split by the split by the Clone or Iterate mediator.

Refer for more information :

Clone Mediator — https://docs.wso2.com/enterprise-integrator/Aggregate+Mediator
Aggregate Mediator — https://docs.wso2.com/enterprise-integrator/Clone+Mediator

Following is a small example to depict the use cases of the Clone and Aggregate Mediators.

1. Consider that there are two back-end services (mocking these using mocky.io)
   - Get account information for a customer
   - Get loans information for a customer

2. Build ESB API with GET resource to input account number.

<resource methods=”GET” uri-template=”/getDetails/{customer_Id}”>

3. Call both above services in parallel and respond back with both account and loans information for the account.

Now lets head to the solution,

Below is the source which creates a Clone mediator and call two services in parallel. It is important to set the id property to the clone tag which we will use to connect with the Aggregate mediator in the latter part of the code.

<log level="custom">
                <property name="LOG" value="Calling GetCustomerDetails API *********"/>
            </log>
            <clone id="RESPONSE">
                <target>
                    <sequence>
                        <call>
                            <endpoint>
                                <http method="get" uri-template="http://run.mocky.io/v3/51971833-3287-4c25-8e69-79f5ceecb454">
                                    <suspendOnFailure>
                                        <initialDuration>-1</initialDuration>
                                        <progressionFactor>1</progressionFactor>
                                    </suspendOnFailure>
                                    <markForSuspension>
                                        <retriesBeforeSuspension>0</retriesBeforeSuspension>
                                    </markForSuspension>
                                </http>
                            </endpoint>
                        </call>
                    </sequence>
                </target>
                <target>
                    <sequence>
                        <call>
                            <endpoint>
                                <http method="get" uri-template="http://run.mocky.io/v3/c3d12cfd-468d-4ab9-9845-2a12c6bdb112">
                                    <suspendOnFailure>
                                        <initialDuration>-1</initialDuration>
                                        <progressionFactor>1</progressionFactor>
                                    </suspendOnFailure>
                                    <markForSuspension>
                                        <retriesBeforeSuspension>0</retriesBeforeSuspension>
                                    </markForSuspension>
                                </http>
                            </endpoint>
                        </call>
                    </sequence>
                </target>
            </clone>
            <log level="custom">
                <property name="LOG" value="Calling GetCustomerDetails API Ended*********"/>
            </log>

Below is the source which creates a Property mediator and an Aggregate mediator.

<property name="info" scope="default">
                <ns:Info xmlns:ns="http://wso2.com"/>
            </property>
            <aggregate id="RESPONSE">
                <completeCondition>
                    <messageCount max="-1" min="-1"/>
                </completeCondition>
                <onComplete aggregateElementType="root" enclosingElementProperty="info" expression="s11:Body/child::* | s12:Body/child::*" xmlns:m0="http://services.samples" xmlns:s11="http://schemas.xmlsoap.org/soap/envelope/" xmlns:s12="http://www.w3.org/2003/05/soap-envelope">
                    <property name="messageType" scope="axis2" type="STRING" value="application/json"/>
                    <respond/>
                </onComplete>
            </aggregate>

Following is the full code used.

<?xml version="1.0" encoding="UTF-8"?>
<api context="/GetCustomerDetails" name="GetCustomerDetails" xmlns="http://ws.apache.org/ns/synapse">
    <resource methods="GET" uri-template="/getDetails/{customer_Id}">
        <inSequence>
            <log level="custom">
                <property name="LOG" value="Calling GetCustomerDetails API *********"/>
            </log>
            <clone id="RESPONSE">
                <target>
                    <sequence>
                        <call>
                            <endpoint>
                                <http method="get" uri-template="http://run.mocky.io/v3/51971833-3287-4c25-8e69-79f5ceecb454">
                                    <suspendOnFailure>
                                        <initialDuration>-1</initialDuration>
                                        <progressionFactor>1</progressionFactor>
                                    </suspendOnFailure>
                                    <markForSuspension>
                                        <retriesBeforeSuspension>0</retriesBeforeSuspension>
                                    </markForSuspension>
                                </http>
                            </endpoint>
                        </call>
                    </sequence>
                </target>
                <target>
                    <sequence>
                        <call>
                            <endpoint>
                                <http method="get" uri-template="http://run.mocky.io/v3/c3d12cfd-468d-4ab9-9845-2a12c6bdb112">
                                    <suspendOnFailure>
                                        <initialDuration>-1</initialDuration>
                                        <progressionFactor>1</progressionFactor>
                                    </suspendOnFailure>
                                    <markForSuspension>
                                        <retriesBeforeSuspension>0</retriesBeforeSuspension>
                                    </markForSuspension>
                                </http>
                            </endpoint>
                        </call>
                    </sequence>
                </target>
            </clone>
            <log level="custom">
                <property name="LOG" value="Calling GetCustomerDetails API Ended*********"/>
            </log>
            <property name="info" scope="default">
                <ns:Info xmlns:ns="http://wso2.com"/>
            </property>
            <aggregate id="RESPONSE">
                <completeCondition>
                    <messageCount max="-1" min="-1"/>
                </completeCondition>
                <onComplete aggregateElementType="root" enclosingElementProperty="info" expression="s11:Body/child::* | s12:Body/child::*" xmlns:m0="http://services.samples" xmlns:s11="http://schemas.xmlsoap.org/soap/envelope/" xmlns:s12="http://www.w3.org/2003/05/soap-envelope">
                    <property name="messageType" scope="axis2" type="STRING" value="application/json"/>
                    <respond/>
                </onComplete>
            </aggregate>
        </inSequence>
        <outSequence>
            <log level="custom">
                <property name="LOG" value="Outsequence*********"/>
            </log>
        </outSequence>
        <faultSequence/>
    </resource>
</api>

Following is the design in integration studio