A typical messaging requirement is to aggregate multiple related messages for processing within a single BPEL process instance. There are two parts to this recipe; the first is to route related messages through to the same instance of a BPEL process. This can be achieved using a correlation set defined against a common value present in each message.
The second is to determine when we have all the messages that belong to the aggregation. Typically, most use cases fall into two broad patterns:
- Fixed duration: In this scenario, we don't know how many messages we expect to receive, so we will process all those received within a specified period of time.
- Wait for all: In this scenario, we know how many messages we expect to receive. Once they have been received, we can process them as an aggregated message. It's usual to combine this with a timeout, so the process doesn't wait forever, if some messages aren't received.
An example of the first pattern is an order aggregation process, whereby we aggregate the orders we receive for a particular book over a period of time (for example, 24 hours) and then place a single order with the publisher for the total amount of orders received.
This recipe makes use of the new aggregation feature in Oracle SOA Suite 11gR1 Patch Set 5 (11.1.1.6.0), so you need to ensure that you have installed either this or a later release of the Oracle SOA Suite.
Create an SOA application with a project containing an empty composite (named WarehouseService in the following example).
- Drag a BPEL process from the SOA Component Palette onto our composite. This will launch the Create BPEL Process wizard. Specify an appropriate name (WarehouseService, in the following screenshot), and for the template, select Base on a WSDL.
- Click on the Find Existing WSDLs icon. This will launch the SOA Resource Browser. Browse the filesystem to select the WSDL that you wish to import and click on OK.
For the purpose of following this example, select the
WarehouseService_1.0.wsdlnode included with the samples for this chapter.Ensure Warehouse Service is selected as the Port Type, and click on OK. This will add BPEL Process Warehouse Service to our composite.
- In the SOA Composite Editor, select the BPEL process service component, as shown next:
- In the Property Inspector pane in the lower-right corner of Oracle JDeveloper, click on the Add icon (circled in the following screenshot).
- This will open the Create Property dialog. In the Name field, enter
bpel.config.reenableAggregationOnComplete, and in the Value field, entertrue; then, click on OK. - Next, we need to create and initialize a
Whileloop to process our messages. Create an xsd:boolean variable namedorderCompleteand use an Assign activity to set it tofalse().Create a variable named
waitUntil, of type xsd:dateTime, and use an Assign activity to set it to the following value:xp20:add-dayTimeDuration-to-dateTime(xp20:current-dateTime(), 'PT1H')
- Next, drag a While activity onto the BPEL process (after the Assign activity) and set its
loopcondition to the following value:$orderComplete = false()
- Then, drag a Pick activity onto the While activity, double-click on the OnMessage icon to open the Edit OnMessage window.
For Partner Link, select the same partner link used by the initial receive activity (
warehouseservice_client, in this example), and select the same operation (submitBookOrder).Click on the auto-create variable (the plus icon) to launch the Create Variable window and give the variable a meaningful name (for example,
nextBookOrder). Next, click on OK. - Now we need to add a step to aggregate the book orders. Drag an Assign activity onto the OnMessage activity.
Add an expression to increment the quantity of the original order received, by the quantity of the new order; in other words:
origOrder/quantity = origOrder/quantity + nextOrder/quantity
The XPath expression, should look something like the following:
$inputVariable.payload.ns2:bookOrder/ns2:quantity + $nextBookOrder.payload.ns2:bookOrder/ns2:quantity
- Select the Pick activity and click on the Add onAlarm icon. Double-click on the onAlarm icon to open the Edit onAlarm window.
Set the first radio button to Until, and set the second radio button to Expression; then, set the expression value to be
$waitUntil. - Next, drag an Assign activity onto the OnAlarm activity, and assign the value
true()to$orderComplete. - Within the Structure view for the BPEL process, right-click on the Properties folder and select Create Property…, as shown in the following screenshot:
This will launch the Create CorrelationSet Property window. Give the property a meaningful name (for example,
isbn), and then click on the search icon to launch the Type Chooser window and select the appropriate schema type (for example,xsd:string). - Click on the Create Property Alias… icon, circled in the following screenshot:
This will open the Property Alias window. In the Type Explorer window, expand the
WarehouseService_1.0.wsdlnode (under the Project WSDL Files folder). Next, expand the Message Types folder and select Part - payload (undersubmitBookOrder) as highlighted in the following screenshot:
- In the Query field, enter the XPath location of the ISBN in the
submitBookOrdermessage, which is:/tns: ns1:bookOrder/ns1:isbn
Then click on OK.
- Within the Structure view for the BPEL process, expand the Correlation Sets folder, and then expand the Process folder. Then, right-click on the Correlation Sets folder and select Create Correlation Set…, as shown in the following screenshot:
This will launch the Create Correlation Set window; give the correlation set a meaningful name, for example,
isbnCS.
- Next, click on the plus icon to launch the Property Chooser window, and select the
isbnproperty created in step 12. - Now we need to initialize the correlation set. Within the BPEL Editor, double-click on the receiveInput activity to open the Edit Receive window, and select the Correlations tab.
- Click on Create Correlation… ( the plus icon circled in the following screenshot). This will add an empty correlation to the Receive activity.
- For the Correlation Set field, select isbnCS from the drop-down list. Next, select Yes from the Initiate field dropdown, as shown in the following screenshot:
- Within the BPEL Editor, double-click on the onMessage activity to open the Edit onMessage window, and select the Correlations tab.
- Click on Create Correlation…, the plus icon. This will add an empty correlation to the onMessage activity. For the Correlation Set field, select isbnCS from the drop-down list. Next, select No from the Initiate dropdown.
- Deploy the
WarehouseServicecomposite to the Oracle SOA Suite and use Enterprise Manager to submit multiplesubmitBookOrdermessages.For book orders that contain the same ISBN number, you should see that they are routed through to the same instance of the BPEL process.
Since the 11.1.1.6 release of the Oracle SOA Suite, Oracle BPEL Process Manager has supported a message aggregation feature. When multiple messages are routed to the same process, the first message is routed to create a new instance and subsequent messages can be routed to continue the created instance using a mid-process receive activity.
By default, this feature is disabled. To enable it, we need to set the property bpel.config.reenableAggregationOnComplete to true.
Once we have enabled aggregation, we still have to define the key that the BPEL engine should use in order to aggregate messages. For this, we use a correlation set (isbnCS, in the previous example) that consists of one or more properties (isbn in our case). These properties are then mapped, using a property alias, to the corresponding field in the messages that are being aggregated.
The combined value of these properties at runtime should result in a unique value (at least, unique across all instances of the same process) that allows the BPEL engine to route the message to the appropriate instance of a process.
On receipt of the first message in the aggregation, we need to tell the BPEL process to initialize the correlation set, which we did by setting the initiate property to true (in step 19). This then tells the BPEL engine to route all other messages that contain the same value to this instance of the BPEL process.