What is Aggregator Connector?
As the name suggests, it means it Aggregates/Accumulates a set of data. Aggregation can be further achieved in 3 different ways:
- Size-Based Aggregator
- Group-Based Aggregator
- Time-Based Aggregator
Let’s start exploring each of these functionalities in detail with a demo.
1. Size-Based Aggregator
This enables you to aggregate the incoming request (entire payload or any specific element) to the size defined in the aggregator configuration.
For example, if you want to bind the incoming request in the batch of 3, then we will set the size limit to 3 in aggregator configuration. Let’s see the implementation now.
- Create a Flow and drag an HTTP Listener and Size-based aggregator connector inside it.
- In the Size-based aggregator connector there are 2 sections as below:
- Incremental Aggregation: This basically keeps track of each current incremental value.
- Aggregation Complete: This section gets triggered once the aggregation condition is satisfied, in this case, the max size. In this, you can either use Flow reference or can use Aggregator Listener for the further execution process.
- Below is the configuration for the Mule flow.
Size-Based Aggregator Flow
Aggregator Listener Flow
- Below is the Connector Configuration.
Size-Based Aggregator Configuration
Aggregator Listener Configuration
Now run the code and send 3 HTTP Requests. In the console, you will see an aggregated output.
**Incremental Aggregation Output**
INFO 2022-01-24 14:02:07,202 [[MuleRuntime].uber.08: [aggregatordemo].aggregatordemoFlow.CPU_LITE @2f2eb4b3] [processor: aggregatordemoFlow/processors/0/route/0/processors/0; event: 1d95e200-7cf0-11ec-a0e1-00155df07a06] org.mule.runtime.core.internal.processor.LoggerMessageProcessor: [TypedValue[value: '[[email protected]', dataType: 'SimpleDataType{type=[B, mimeType="application/json; charset=UTF-8"}']]
INFO 2022-01-24 14:02:11,755 [[MuleRuntime].uber.08: [aggregatordemo].aggregatordemoFlow.CPU_LITE @2f2eb4b3] [processor: aggregatordemoFlow/processors/0/route/0/processors/0; event: 204c7680-7cf0-11ec-a0e1-00155df07a06] org.mule.runtime.core.internal.processor.LoggerMessageProcessor: [TypedValue[value: '[[email protected]', dataType: 'SimpleDataType{type=[B, mimeType="application/json; charset=UTF-8"}'], TypedValue[value: '[[email protected]', dataType: 'SimpleDataType{type=[B, mimeType="application/json; charset=UTF-8"}']]
INFO 2022-01-24 14:02:15,460 [[MuleRuntime].uber.08: [aggregatordemo].aggregatordemoFlow.CPU_LITE @2f2eb4b3] [processor: aggregatordemoFlow/processors/0/route/1/processors/0; event: 2281a600-7cf0-11ec-a0e1-00155df07a06] org.mule.runtime.core.internal.processor.LoggerMessageProcessor: [TypedValue[value: '[[email protected]', dataType: 'SimpleDataType{type=[B, mimeType="application/json; charset=UTF-8"}'], TypedValue[value: '[[email protected]', dataType: 'SimpleDataType{type=[B, mimeType="application/json; charset=UTF-8"}'], TypedValue[value: '[[email protected]', dataType: 'SimpleDataType{type=[B, mimeType="application/json; charset=UTF-8"}']]
**Aggregation Complete Output**
INFO 2022-01-24 14:02:15,462 [[MuleRuntime].uber.05: [aggregatordemo].aggregatordemoFlow1.CPU_INTENSIVE @58aef977] [processor: aggregatordemoFlow1/processors/1; event: 1787be10-7cf0-11ec-a0e1-00155df07a06] org.mule.runtime.core.internal.processor.LoggerMessageProcessor: [
{
"Text": 1
},
{
"Text": 2
},
{
"Text": 3
}
]
2. Group-Based Aggregator
This enables you to aggregate the data (entire payload or any specific element) into groups based on the Group id and the batch size defined in the configuration.
For example, if you want to group employees based on their status as active or inactive we can make use of this. Let’s see the implementation now.
- Create a Flow and drag an HTTP Listener and Group-based aggregator connector inside it.
- Now configure Name, Group id, and Group size in Group-based aggregator connector depending on your requirement.
- Eviction Time is the time till it remembers the Group id, the default is 180 secs.
- The rest of the configuration remains the same as above.
Group-Based Aggregator Configuration
Now run the code and send Employee data with status as Active and Inactive and you can see the output being grouped with respective statuses.
**Incremental Aggregation Output**
INFO 2022-01-24 14:50:37,833 [[MuleRuntime].uber.05: [aggregatordemo].aggregatordemoFlow.CPU_LITE @689874a9] [processor: aggregatordemoFlow/processors/0/route/0/processors/0; event: e4742a20-7cf6-11ec-b9da-00155df07a06] org.mule.runtime.core.internal.processor.LoggerMessageProcessor:
org.mule.runtime.core.internal.message.DefaultMessageBuilder$MessageImplementation
{
payload=[TypedValue[value: '[[email protected]', dataType: 'SimpleDataType{type=[B, mimeType="application/json; charset=UTF-8"}']]
mediaType=*/*
att[email protected]217e7427
attributesMediaType=*/*
}
INFO 2022-01-24 14:50:44,767 [[MuleRuntime].uber.05: [aggregatordemo].aggregatordemoFlow.CPU_LITE @689874a9] [processor: aggregatordemoFlow/processors/0/route/0/processors/0; event: e89683a0-7cf6-11ec-b9da-00155df07a06] org.mule.runtime.core.internal.processor.LoggerMessageProcessor:
org.mule.runtime.core.internal.message.DefaultMessageBuilder$MessageImplementation
{
payload=[TypedValue[value: '[[email protected]', dataType: 'SimpleDataType{type=[B, mimeType="application/json; charset=UTF-8"}'], TypedValue[value: '[[email protected]', dataType: 'SimpleDataType{type=[B, mimeType="application/json; charset=UTF-8"}']]
mediaType=*/*
att[email protected]2064cc7e
attributesMediaType=*/*
}
INFO 2022-01-24 14:50:50,556 [[MuleRuntime].uber.05: [aggregatordemo].aggregatordemoFlow.CPU_LITE @689874a9] [processor: aggregatordemoFlow/processors/0/route/1/processors/0; event: ec09ff80-7cf6-11ec-b9da-00155df07a06] org.mule.runtime.core.internal.processor.LoggerMessageProcessor:
org.mule.runtime.core.internal.message.DefaultMessageBuilder$MessageImplementation
{
payload=[TypedValue[value: '[[email protected]', dataType: 'SimpleDataType{type=[B, mimeType="application/json; charset=UTF-8"}'], TypedValue[value: '[[email protected]', dataType: 'SimpleDataType{type=[B, mimeType="application/json; charset=UTF-8"}'], TypedValue[value: '[[email protected]', dataType: 'SimpleDataType{type=[B, mimeType="application/json; charset=UTF-8"}']]
mediaType=*/*
att[email protected]623f8ee0
attributesMediaType=*/*
}
**Aggregation Complete Output**
INFO 2022-01-24 14:50:50,561 [[MuleRuntime].uber.03: [aggregatordemo].aggregatordemoFlow1.CPU_INTENSIVE @61e80f72] [processor: aggregatordemoFlow1/processors/1; event: Active] org.mule.runtime.core.internal.processor.LoggerMessageProcessor: [
{
"name": "ABC",
"status": "Active"
},
{
"name": "MNO",
"status": "Active"
},
{
"name": "XYZ",
"status": "Active"
}
]
Since we have set eviction time as 180 seconds we cannot send another request with the status Active till 180 seconds gets over. You will receive the below error in the console.
ERROR 2022-01-24 14:52:03,275 [[MuleRuntime].uber.05: [aggregatordemo].aggregatordemoFlow.CPU_LITE @689874a9] [processor: aggregatordemoFlow/processors/0; event: 1761bb50-7cf7-11ec-b9da-00155df07a06] org.mule.runtime.core.internal.exception.OnErrorPropagateHandler:
********************************************************************************
Message : Trying to aggregate a new element to the group with id: Active ,but it's already complete
Element : aggregatordemoFlow/processors/0 @ aggregatordemo:aggregatordemo.xml:14 (Group based aggregator)
Element DSL : <aggregators:group-based-aggregator doc:name="Group based aggregator" doc:id="dee44800-5b56-4f64-bb0a-ec2336df6334" name="groupAggregator" groupId="#[payload.status]" groupSize="3">
<aggregators:incremental-aggregation>
<logger level="INFO" doc:name="Logger" doc:id="f05a8637-8422-4764-b58b-f81e2b6455ed"></logger>
</aggregators:incremental-aggregation>
<aggregators:aggregation-complete>
<logger level="INFO" doc:name="Logger" doc:id="b99e2e44-21ff-4d83-a00f-4c6a94533b4f"></logger>
</aggregators:aggregation-complete>
</aggregators:group-based-aggregator>
Error type : AGGREGATORS:GROUP_COMPLETED
FlowStack : at aggregatordemoFlow(aggregatordemoFlow/processors/0 @ aggregatordemo:aggregatordemo.xml:14 (Group based aggregator))
(set debug level logging or '-Dmule.verbose.exceptions=true' for everything)
********************************************************************************3. Time-Based Aggregator
This enables you to aggregate the request (entire payload or any specific element) for the time period defined in the aggregator configuration.
For example, if you want to bind all the incoming requests for a time period of 30 sec you need to set this time period in a Time-based aggregator configuration. You can also restrict the size by defining the Max size limit. Let’s see the implementation now.
- Create a Flow and drag an HTTP Listener and Time-based aggregator connector inside it.
- Now configure Time Period and Max Size in Time-based aggregator connector depending on your requirement.
- The rest of the configuration remains the same as above.
Time-Based Aggregator Configuration
Now run the code and send 2-3 requests within the time specified and then you can see the aggregated response in the output.
INFO 2022-01-24 15:28:20,908 [[MuleRuntime].uber.03: [aggregatordemo].aggregatordemoFlow.CPU_LITE @61cf581] [processor: aggregatordemoFlow/processors/0/route/0/processors/0; event: 294037c0-7cfc-11ec-91f8-00155df07a06] org.mule.runtime.core.internal.processor.LoggerMessageProcessor: [TypedValue[value: '[[email protected]', dataType: 'SimpleDataType{type=[B, mimeType="application/json; charset=UTF-8"}']]
INFO 2022-01-24 15:28:21,873 [[MuleRuntime].uber.03: [aggregatordemo].aggregatordemoFlow.CPU_LITE @61cf581] [processor: aggregatordemoFlow/processors/0/route/0/processors/0; event: 29ed8ec0-7cfc-11ec-91f8-00155df07a06] org.mule.runtime.core.internal.processor.LoggerMessageProcessor: [TypedValue[value: '[[email protected]', dataType: 'SimpleDataType{type=[B, mimeType="application/json; charset=UTF-8"}'], TypedValue[value: '[[email protected]', dataType: 'SimpleDataType{type=[B, mimeType="application/json; charset=UTF-8"}']]
INFO 2022-01-24 15:28:22,859 [[MuleRuntime].uber.03: [aggregatordemo].aggregatordemoFlow.CPU_LITE @61cf581] [processor: aggregatordemoFlow/processors/0/route/0/processors/0; event: 2a83db50-7cfc-11ec-91f8-00155df07a06] org.mule.runtime.core.internal.processor.LoggerMessageProcessor: [TypedValue[value: '[[email protected]', dataType: 'SimpleDataType{type=[B, mimeType="application/json; charset=UTF-8"}'], TypedValue[value: '[[email protected]', dataType: 'SimpleDataType{type=[B, mimeType="application/json; charset=UTF-8"}'], TypedValue[value: '[[email protected]', dataType: 'SimpleDataType{type=[B, mimeType="application/json; charset=UTF-8"}']]
INFO 2022-01-24 15:28:40,878 [[MuleRuntime].uber.05: [aggregatordemo].aggregatordemoFlow1.CPU_INTENSIVE @31dd2e47] [processor: aggregatordemoFlow1/processors/1; event: 294dcc50-7cfc-11ec-91f8-00155df07a06] org.mule.runtime.core.internal.processor.LoggerMessageProcessor: [
{
"Text": 1
},
{
"Text": 1
},
{
"Text": 1
}
]
