Aggregator Pattern in NWBPM

Dear SCN users,

This blog gives the step by step approach of collecting messages based on correlation and count(Aggregator pattern).

Prerequisties:

1.NWDS Installation.

2.PI 7.31 and above with NWBPM installation

In this scenario I am collecting messages based on product and number of messages collected will be 3(count = 3)

ESR Objects:

Below is the MT Type I have created.

Input:

Output:

Below is the MT type for Output

Based on the product Brand Messages will be collected(For Example all LG Products will be collected together)

1.we need to create 4 service interfaces as shown below

Outbound Interface for sending data to PI,Refering to Input Message type

Inbound Interface to send message to NWBPM.Here the pattern should be XI 3.0 Stateless and Refering to Input Message type

outbound Interface to send message from NWBPM to PI.Here also pattern should be Statelss XI 3.0 compatitable and Refering to Output Message type

Inbound Interface to send message from PI to File system Refering to Output Message type

NWDS Steps:

1.Open NWDS->Select perspective->development

2.Then new project,and select lanes and name them

3. And service interfaces should be imported from PI.To import service interfaces right click on service interface/mapping and import them from PI.

Below is the pattern for collecting based on count and correlaction.

Here 4 data artifcats are used

1 is input message artifact

2 is Output message artifact

3 count- to keep the count of messages

4 Product- to keep the correlation Id(product)

Start Message:

Here the Event trigger will be the start which will refer to Inbound Service Iterface.

In the output mapping start Event Incoming message is mapped to the Input data artifact as shown below

And count is intialised with the value of zero

And product is assigned with the product Brand(based on which messages will be collected)-correlation.

Count Event(Automated activity):

In the count Event,the count will increment based on the collected messages.Here automated activity or mapping activity can be used.

Here I selected Sampletimer as interface which is not going to impact anything.And in the Output mapping Count is incremented using add Function

Intermediate Message:

Here the messages are collected based on correlation.

The event trigger here will be same as that of start.

In the Output mapping ,Input message is mapped to the input data artifact using append assignment

And in the correlation condition the Incoming product brand is compared with product assingned in start event.If the condition is true,messages are collected.

Exclusive Choice Event:

This is used to check the condtion and loop accordingly.

If the count<3,it is looped to Intermediate message ,If the count=3 then mapping event is looped.

Mapping Event(Automated Activity):

In mapping event,the mapping program as interface is selected.

1.From Input artifact to the input of the mapping in the input mapping tab

2.From output of the mapping to the output artifact in the output mapping tab

Message Sent(Automated Activity):

This activity is automated activity which is used to send the message to PI.

Here BPM Outbound interface is selected as interface

And in the input mapping ,from the output artifact message is passed to the interface.

Service References:

For mapping and message sent event service refernces should be configured.

For Mapping event,select new service reference group and select local host option

For Message event select new service reference group and select XI details and fill details as shown below.

Integration Directory Objects:

In ID 2 Icos need to be configured.

One for sending message to BPM

Below is the Soap receiver channel to send message to BPM

In second ICO,message is sent from BPM to PI to IFile interface.

Testing:

I am sending 6 messages corresponding to 2 different product brands,3 are correlated collected as shown below