Scatter-Gather or Broadcast Aggregate is a popular Enterprise Integration Pattern (EIP). The example given in the EIP catalog talks about a scenario where you send a quote request to a 3 vendors, and get quotes for certain items, and send back the response from the vendor who provides the best price.
We will show how this can be simulated using the WSO2 ESB, , which is the fastest, proven & best performing & most flexible open source ESB. For the sake of convenience, we will take the SimpleStockQuoteService as the service provided by the vendors, and return the best quote to the client.
We will use 3 constructs in the ESB to implement this scenario.
We will write a simple proxy service called ScatterGatherProxy. The assumption here is that all 3 vendors implement the same service contract. If the service contracts are different, we will first have to do the appropriate transformations before sending the requests to the vendor services, and then transform back the responses. The XSLT mediator can be used to do that. The configuration of this proxy service is shown below.
<?xml version="1.0" encoding="UTF-8"?>
<proxy xmlns="http://ws.apache.org/ns/synapse" name="ScatterGatherProxy" transports="https http" startOnLoad="true" trace="disable">
<description/>
<target>
<inSequence>
<clone>
<target>
<endpoint name="vendorA">
<address uri="http://localhost:9000/services/SimpleStockQuoteService/"/>
</endpoint>
</target>
<target>
<endpoint name="vendorB">
<address uri="http://localhost:9001/services/SimpleStockQuoteService/"/>
</endpoint>
</target>
<target>
<endpoint name="vendorC">
<address uri="http://localhost:9002/services/SimpleStockQuoteService/"/>
</endpoint>
</target>
</clone>
</inSequence>
<outSequence>
<log level="full"/>
<aggregate>
<completeCondition>
<messageCount min="3"/>
</completeCondition>
<onComplete xmlns:m1="http://services.samples/xsd" xmlns:m0="http://services.samples" expression="//m0:return">
<enrich>
<source xmlns:m1="http://services.samples/xsd" clone="true" xpath="//m0:return[not(preceding-sibling::m0:return/m1:last <= m1:last) and not(following-sibling::m0:return/m1:last < m1:last)]"/>
<target type="body"/>
</enrich>
<send/>
</onComplete>
</aggregate>
</outSequence>
</target>
</proxy>
To be able to run this sample, you will need to deploy the SimpleStockQuoteService on the Sample AxisServer which ships with the WSO2 ESB, and start 3 instances of it, with HTTP ports 9000, 9001 & 9002 respectively.
Next deploy the above proxy service on WSO2 ESB.
Now let's use soapUI to send the following request to the ScatterGatherProxy service.
<soapenv:Envelope xmlns:soapenv="http://schemas.xmlsoap.org/soap/envelope/" xmlns:ser="http://services.samples">
<soapenv:Header/>
<soapenv:Body>
<ser:getSimpleQuote>
<ser:symbol>IBM</ser:symbol>
</ser:getSimpleQuote>
</soapenv:Body>
</soapenv:Envelope>
Since we have enabled the log mediator inside the outSequence, you will see 3 responses from the 3 vendors. The logs you will see will be similar to the following
In soapUI, you will get the response from the vendor who provided the best quote.
If you carefully compare the logged response messages & the response received by the client, you will notice that the ScatterGatherProxy service returns the best quote to the client (soapUI)
Underneath the Hood
So what is going on in this proxy service?Let us first look at the inSequence of the ScatterGatherProxy service. In the inSequence, we are simply cloning, or making 3 copies of the request that was sent by the client using the Clone mediator, and forwarding those requests to the 3 vendor services (SimpleStockeQuoteService).
The responses to those 3 requests will be received at the outSequence. Here we first log all the received responses, and then use the Aggregate mediator to aggregate the responses. The onComplete sequence of the Aggregate mediator will get called once all 3 responses are received or the specified completion condition is met. We aggregate the responses based on the value of the "return" element in the response. Next we use the Enrich mediator to extract the response which contains the best quote. We use the following XPath 1.0 expression to do this;
//m0:return[not(preceding-sibling::m0:return/m1:last <= m1:last) and not(following-sibling::m0:return/m1:last < m1:last)]
This is a complex way of saying, "pick the response which has the lowest 'last' value". Unfortunately, XPath 1.0, which is the default supported by WSO2 ESB, does not support the XPath 2.0 min function which can reduce the complexity of the above statement by a great deal. Once we find the proper response, we enrich the SOAP body with that response, and then send that response back to the client (soapUI)
