Mule Aggregator Connector

In this article, we will explore and understand the use of the Mule Aggregator Connector and demonstrate 3 of its core functionalities.

Karan Gupta

Jan. 30, 22 · Tutorial

Like (3)

Save

7.6K Views

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.

     Plain Text 
   
 
 
   **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: '[B@63d6397f', 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: '[B@46c24667', dataType: 'SimpleDataType{type=[B, mimeType='application/json; charset=UTF-8'}'], TypedValue[value: '[B@bd58c3c', 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: '[B@4a986e9', dataType: 'SimpleDataType{type=[B, mimeType='application/json; charset=UTF-8'}'], TypedValue[value: '[B@6326951f', dataType: 'SimpleDataType{type=[B, mimeType='application/json; charset=UTF-8'}'], TypedValue[value: '[B@b79f9dd', 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.

     Plain Text 
   
 
 
   **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: '[B@4e7a5601', dataType: 'SimpleDataType{type=[B, mimeType='application/json; charset=UTF-8'}']]
  mediaType=*/*
  attributes=org.mule.extension.aggregator.api.AggregationAttributes@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: '[B@51e94f99', dataType: 'SimpleDataType{type=[B, mimeType='application/json; charset=UTF-8'}'], TypedValue[value: '[B@2e7fcd2d', dataType: 'SimpleDataType{type=[B, mimeType='application/json; charset=UTF-8'}']]
  mediaType=*/*
  attributes=org.mule.extension.aggregator.api.AggregationAttributes@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: '[B@4b29977e', dataType: 'SimpleDataType{type=[B, mimeType='application/json; charset=UTF-8'}'], TypedValue[value: '[B@3e61fe9e', dataType: 'SimpleDataType{type=[B, mimeType='application/json; charset=UTF-8'}'], TypedValue[value: '[B@3e104949', dataType: 'SimpleDataType{type=[B, mimeType='application/json; charset=UTF-8'}']]
  mediaType=*/*
  attributes=org.mule.extension.aggregator.api.AggregationAttributes@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.

     Plain Text 
   
 
 
   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.

     Plain Text 
   
 
 
   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: '[B@502dd49e', 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: '[B@7d50b4f4', dataType: 'SimpleDataType{type=[B, mimeType='application/json; charset=UTF-8'}'], TypedValue[value: '[B@42282847', 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: '[B@56bd5e36', dataType: 'SimpleDataType{type=[B, mimeType='application/json; charset=UTF-8'}'], TypedValue[value: '[B@c2bfde6', dataType: 'SimpleDataType{type=[B, mimeType='application/json; charset=UTF-8'}'], TypedValue[value: '[B@4e655dbd', 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
  }
]
 
  

Connector (mathematics)

Opinions expressed by DZone contributors are their own.

Related

Trending