Concurrently Processing Split Payloads and Collecting Result in MuleSoft

As a MuleSoft Certified Architect, Designer, and Developer, I recently worked on API implementations for one of our clients using MuleSoft’s CloudHub. One feature that we use in some APIs implementations is to split an input payload into multiple fragments, process each payload unit in parallel, and then merge the responses of each payload unit to send a final consolidated response.

In this article, I will demonstrate splitting a Mule payload into multiple fragments, processing each fragment in parallel, and finally, collecting the output of each fragment into a final response to be sent back to the consumer using MuleSoft.

As we know, Mule provides the Request-Reply scope to embed a "pocket" of asynchronous processing within a Mule flow. Since we are talking about parallel processing and sending a final response, we will use the Request-Reply scope in Mule. To split and join the payload, we will use the MuleSoft Splitter and Aggregator component, respectively.

Let’s get started and see how we can use these components to achieve Split-Join functionality with parallel processing in MuleSoft.

1. Setting Request and Reply Scope

The very first step we will take is to set up a request and reply scope of Mule and use VM transport to separate the Mule message processing thread.

As we can see in the above screen, the Input message coming to the Mule flow will be sent to an outbound VM queue, which submits an asynchronous request to another flow. The response VM will receive an asynchronous response from another flow.

2. Splitting Mule Payload

We will use the Mule Splitter component to split our Mule payload since I am working with an XML payload; I have used an XPATH expression to split my Mule payload.

As you can see in the above screen, I have used one variable component to store the original payload before splitting. My requirement is to call an external web service with each split fragment of the input payload. I have used Mule DataWeave to transform each payload fragment according to the external web service format to make a successful call. This transformation uses values from the original payload, which I have used from flowVar defined before the split component. The final payload is again being posted to a VM queue which will be processed on a separate thread to achieve parallel processing by running on different threads.

3. Calling an External Web Service With the Split Payload

Now let’s make an outbound call to the external web service for each split payload and get the response in a VM queue.

As you can see in the above screen, I have wrapped the external web service call inside the message enricher but mapped the payloads. This has been done to keep MULE_CORRELATION_GROUP_SIZE, MULE_CORRELATION_SEQUENCE, and MULE_CORRELATION_ID inbound properties, which the splitter generates for each payload unit. We have to keep these properties in the flow to use the collection aggregator to collect the final output.

4. Aggregating Final Responses

Now, we have responses from the external web service call for each payload unit in the VM which need to be aggregated.

As shown above, we have used the collection aggregator component to collect the responses for each payload unit in one payload and do the transformation to prepare the format we want the final response for the consumer. The final transformed payload has been published to the VM queue to which our main flow request-reply scope response VM is listening.

In conclusion, we received a Mule payload from the consumer with a repetitive block. We split each repetitive block using an XPATH expression and did a transformation to call an external web service. The response for each payload unit has been aggregated using the collection aggregator and sent back to the consumer in its defined format. The best part is that we have processed all of these steps in separate threads, and they worked in parallel.

Let’s share our knowledge to expand our MuleSoft community.

Thank you!