IBM Integration Bus: IIB Aggregate nodes sample with HTTP web services

In this tutorial, we’ll explore an IIB Aggregate nodes sample with HTTP web services. Aggregation is a scenario where we receive single input message, derive and generate multiple related requests and fan-out the same to back-end. Thereafter, we fan-in the corresponding replies to produce a single aggregated reply message. As per IBM definition, Aggregation is an extension of the request/reply application model. It combines the generation and concurrent fan-out of a number of related requests with the fan-in of the corresponding replies, and compiles those replies into a single aggregated reply message. IIB as a product makes it easy to implement complex integration scenario with Aggregation support.

There are few articles (for example, here and here) in IBM site that explain the usage of aggregate nodes but all of them use additional MQ queues to support aggregation. In this tutorial, my goal is to demonstrate the aggregation of HTTP(s) based webservices without using any additional MQ queues. In this demonstration, I’ll also add further complexity where you won’t know the number of time you need to call back-end services beforehand; instead you need to look at the input data to determine how many times you need to call the back-end.

IIB Aggregation node

IBM Integration Bus provides three message flow nodes to support aggregation:

• AggregateControl node – It marks the beginning of a fan-out of requests that are part of an aggregation.
• AggregateRequest node – It records the fact that the request messages have been sent. It also collects information that helps the AggregateReply node to construct the aggregated reply message.
• AggregateReply node – It marks the end of an aggregation fan-in. It collects replies and combines them into a single aggregated reply message. This node creates the LocalEnvironment.ComIbmAggregateControlNode folder that contains aggregated responses along with many other information.

IIB Aggregation Queues

The above aggregation nodes store state for aggregations on WebSphere MQ queues. By default, the following storage queues are used:

• SYSTEM.BROKER.AGGR.CONTROL
• SYSTEM.BROKER.AGGR.REPLY
• SYSTEM.BROKER.AGGR.REQUEST
• SYSTEM.BROKER.AGGR.UNKNOWN
• SYSTEM.BROKER.AGGR.TIMEOUT

We can also create alternative storage queues as well. This is the reason, for IIB to execute aggregation scenario, the integration node MUST have an associated MQ Queue Manager.

Tech Stack

• Windows 10 Professional
• IIB 10.0.0.8 (Developer edition)
• WMQ 9 (Developer edition)

Scenario

An application wants to create a dashboard with customer ids and their names, phone numbers. The application has access tot he customer id but wants to retrieve the name and phone number from a back-end system.

The back-end system already has a legacy web-service that accepts one customer-id at a time and responds back with the corresponding details. But consuming application does not want to call it multiple times. Hence, we have a requirement to build a service in IIB that accepts a bunch to customer id from the consuming application and aggregates the customer details by calling the back-end application for each customer individually. IIB then responds back to consuming application with a consolidated details. In order to implement this scenario –

• We’ll build one message flow that receives the input with multiple customer id, generates one request for each of the ids and invokes the back-end service. It also collates each of the responses from back-end and responds back to the the consuming app with an aggregated reply.
• One messages flow that works as legacy back-end service that recievs receive one customer id and responds back with corresponding details.

Messages used in the sample

Input Message from Consumer

{
“CustomerInfoReq”: [
{
“custId”: “cust1”
},
{
“custId”: “cust2”
},
{
“custId”: “cust3”
}
]
}

Output Message to Consumer

[{“customer”: [
{
“customerId”: “cust1”,
“name”: “cust1-name”,
“phone”: “cust1-phone”
},
{
“customerId”: “cust2”,
“name”: “cust2-name”,
“phone”: “cust2-phone”
},
{
“customerId”: “cust3”,
“name”: “cust3-name”,
“phone”: “cust3-phone”
}
]}]

Input to legacy back-end service

Path Parameter: /customer/{id}

Output from legacy back-end service

<customerInfo>
<customerId>cust1</customerId>
<name>cust1-name</name>
</customerInfo>

Create an Integration Node

• For aggregation to work, we need to create an Integration Node with an associated queue manager. Hence, first create a QM from MQ explorer or from command line. I have named the QM as IIBQM
• Create a local integration node from either GUI or command console. I have used command console to create a local integration node with the name LOCALNODE and associated IIBQM created in the above step. The command I have used here is –

mqsicreatebroker LOCALNODE -q IIBQM

• I have also created an integration server named default. I used the toolkit for creating it.

Build the legacy back-end REST API

• Open the IIB toolkit, create a new REST API with following info –
  • Name:- LegacyBackendService
  • Select the option that says – Create a REST API and define resources and operations yourself. Then fill as below:
    • API Base Path: /legacybackendservice/v1
    • Version: 1.0.0
  • Click on Finish button.
• The toolkit will create a REST API Project with above information.
• Double click on the REST API Description in Application Development pane. The REST API description window will open up. Here, we’ll create the API resource needed for us.
• Click on the plus icon on the right of Resources section. A new Create Resource window will pop up. Put the resource path as –
  • /customer/{id}
  • Select GET operation
  • Click Apply button
  • With this, a new resource will be created. Change the operation-name and description as below –
    • operation name: – getCustomerInfo
    • operation description:- Retrieve customer Info
  • Save.
• Click on the icon on the right to create an implementation subflow of this newly created operation. Refer to the red encircled icon in the below picture.

• When you click above icon, a new subflow will open. Drag/drop a compute node. The flow will look like –

• Modify the ESQL as follows –

CREATE COMPUTE MODULE getCustomerInfo_Compute
CREATE FUNCTION Main() RETURNS BOOLEAN
BEGIN
CALL CopyMessageHeaders();
DECLARE custid CHAR ”;
SET custid = InputLocalEnvironment.REST.Input.Parameters.id;
Set OutputRoot.XMLNSC.customerInfo.customerId = custid;
Set OutputRoot.XMLNSC.customerInfo.name = custid || ‘-name’;
RETURN TRUE;
END;
CREATE PROCEDURE CopyMessageHeaders() BEGIN DECLARE I INTEGER 1; DECLARE J INTEGER; SET J = CARDINALITY(InputRoot.*[]); WHILE I < J DO SET OutputRoot.*[I] = InputRoot.*[I]; SET I = I + 1; END WHILE; END;
END MODULE;

Build the REST API with Aggregation

• In the IIB toolkit, create a new REST API with following info –
  • Name:- AggregationCustomerAPI
  • Select the option that says – Create a REST API and define resources and operations yourself. Then fill as below:
    • API Base Path: /aggregationcustomerapi/v1
    • Version: 1.0.0
  • Click on Finish button.
• The toolkit will create a REST API Project with above information.
• Double click on the REST API Description in Application Development pane. The REST API description window will open up. Here, we’ll create the API resource needed for us.
• Click on the plus icon on the right of Resources section. A new Create Resource window will pop up. Put the resource path as –
  • /customers
  • Select POST operation
  • Click Apply button
  • With this, a new resource will be created. Change the operation-name and description as below -operation name (NOTE:- I am not following much of REST principles/naming-conventions here. Idea is to demonstrate Aggregation as easily possible 🙂 )
    • Operation Name:- getCustomerDetails
    • Description:- Get Customer Details
  • Save
• Click on the icon on the right to create an implementation subflow of this newly created operation.

• Next, we’ll build the aggregation logic in the newly created subflow. I’ll not explain each node as such. It’ll be pretty much self explanatory when you look at the following picture and the property of each node, I made changes to.

IIB message flow Node configurations

• Aggregate Control

Aggregate Name	AGGR
Timeout (sec)	120

• Compute Node (split)

Transaction	Automatic
ESQL Module	{default}:getCustomerDetails_Split
Compute Mode	LocalEnvironment and Message

CREATE COMPUTE MODULE getCustomerDetails_Split
CREATE FUNCTION Main() RETURNS BOOLEAN
BEGIN
DECLARE I,J INT 1;
DECLARE reqURL CHAR ‘http://localhost:7080/legacybackendservice/v1/customer/’;
DECLARE custId CHAR ”;
DECLARE myref REFERENCE TO InputRoot.JSON.Data.CustomerInfoReq;
SET J = CARDINALITY(myref.Item[]); CALL CopyMessageHeaders(); WHILE I <= J DO SET OutputLocalEnvironment = InputLocalEnvironment; SET OutputLocalEnvironment.Destination.HTTP.RequestURL = reqURL || myref.Item[I].custId; PROPAGATE TO TERMINAL 'out' DELETE NONE; SET I = I + 1; END WHILE; RETURN FALSE; END; CREATE PROCEDURE CopyMessageHeaders() BEGIN DECLARE I INTEGER 1; DECLARE J INTEGER; SET J = CARDINALITY(InputRoot.*[]); WHILE I < J DO SET OutputRoot.*[I] = InputRoot.*[I]; SET I = I + 1; END WHILE; END;
END MODULE;

ESQL module for split compute node

• HTTP Asynchronous Request

Unique Identifier	ASYNCHTTP
Web Service URL	http://localhost:7080/legacybackendservice/v1/customer/ (This is dummy one. Actual URL is being constructed in the previous ESQL module)
Request Timeout (sec)	120
HTTP Method	GET

• Aggregate Request

Folder Name

RESP

• HTTP Asynchronous Response

Unique Identifier	ASYNCHTTP
Message Domain	XMLNSC

• Aggregate Reply

Aggregate Name	AGGR
Unknown Message Timeout (sec)	120

• Compute Node (Merge)

Transaction	Automatic
ESQL Module	{default}:getCustomerDetails_Merge
Compute Mode	Message

CREATE COMPUTE MODULE getCustomerDetails_Merge
CREATE FUNCTION Main() RETURNS BOOLEAN
BEGIN
DECLARE I, J INT 1;
DECLARE myref1 REFERENCE TO InputRoot.ComIbmAggregateReplyBody;
SET J = CARDINALITY(myref1.RESP[]);
CREATE FIELD OutputRoot.JSON.Data IDENTITY (JSON.Array)Data;
WHILE I <= J DO
SET OutputRoot.JSON.Data.CustomerInfoRes.customer[I].customerId=myref1.RESP[I].XMLNSC.customerInfo.customerId;
SET OutputRoot.JSON.Data.CustomerInfoRes.customer[I].name=myref1.RESP[I].XMLNSC.customerInfo.name;
SET OutputRoot.JSON.Data.CustomerInfoRes.customer[I].phone=myref1.RESP[I].XMLNSC.customerInfo.phone;
SET I = I + 1;
END WHILE;
RETURN TRUE; END;
END MODULE;

ESQL module for merge response compute node

Deploy and Test the flow

• Well, now the development is complete and we will drag both the REST API Services to the default integration server of LOCALNODE integration node. This will ensure that both the APIs are deployed.
• NOTE:- The port of the IIB flows depend on your environment. Ensure to cross check that and update the same in the Split compute node accordingly. This is where we are creating the target back-ends service URL and setting in the LocalEnvironment.
• Once successfully deployed, now is the time to test the flow. You can use any tool of your choice. I have tested the API with SOAP UI and following is the result

Conclusion

I hope this tutorial helps you implementing such solution where you want to implement aggregation based on purely web services and you don’t want to use separate MQ queues. Please let me know should you have any question, comment.

Happy learning folks!!