A typical use case with a composite would be submitting a request to an external resource/system that only supports a single connection at a time. In such a case, we need to protect against parallel composite instances submitting concurrent requests to that resource.
This implies that we need some way to serialize requests for that resource (probably on a first-in, first-out basis). A common design pattern for achieving this is the singleton, first documented in the book Design Patterns: Elements of Reusable Object-Oriented Software, Gang of Four (where they use a print spooler as an example).
Now, BPEL doesn't explicitly support the notion of a singleton, however it does allow you to simulate one using a variation of the Message aggregation within a composite recipe, which is good enough for the purpose of what we are trying to achieve.
For the purpose of this recipe, we will extend the PublisherSkynet service used in the Using dynamic partner link recipe, so that we can only publish a single order to Skynet at a time.
We have extended the implementation of this composite for the BPEL process PublisherSkynet to invoke the BPEL process PublisherAsyncService. You can find the code for this extended example in the Getting ready section of this recipe.
For this recipe, we are going to show how to modify the PublisherAsyncService process to act as a singleton. So, you will need to open this composite with JDev to follow this sample recipe.
- In the SOA Composite Editor, ensure you have selected the BPEL process
PublisherAsyncService. In the Property Inspector pane, in the lower-right corner of Oracle JDeveloper, click on the Add icon.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. - Open the BPEL process
PublisherSkynet. Open the Assign activitysetSubmitBookOrder, and add a third copy rule to set the value oftokento the following expression:substring( xp20:current-dateTime(), 1, 16)
Save and close this process.
- Open the BPEL process
PublisherAsyncService. First, we need to create and initialize aWhileloop to process the messages in sequence. Create an xsd:boolean variable named processingComplete, and 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 expression:
xp20:add-dayTimeDuration-to-dateTime( concat($inputVariable.payload/ns1:token, ':00'), 'PT1M')
Next, drag a While activity onto the BPEL process (after the Assign activity) and set its
loopcondition to the following expression:$processingComplete = 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 (
publisherasyncservice_client, in this example), and select the same operation (submitBookOrder).Click on the auto-create variable (plus icon) to launch the Create Variable window. Give the variable a meaningful name (for example,
nextBookOrder). - Next, we need to add the logic to process the message. This is where we would typically call out to an external resource/system that only supports a single connection at a time.
For the purpose of this recipe, we are just going to add a 10 second delay to simulate the time required by the external system to process the request. So, drag a Wait activity onto the onMessage branch, and set it to wait for 10 seconds.
- 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
$waitUntil. - Within the Structure view for the BPEL process, right-click on the
Propertiesfolder and select Create Property…. This will launch the Create Property window. Give the property a meaningful name, (for example,token), and then click on the search icon to launch the Type Chooser window; select the appropriate schema type (for example,xsd:string). - Click on the Create Property Alias… icon. This will open the Property Alias window. In the Type Explorer window, expand the
PublisherAsyncService_1.0.wsdlnode (under the Project WSDL Files folder). Next, expand the Message Types folder and select the Part - payload under thesubmitBookOrdermessage type.In the Query field, enter the XPath location of the token in the
submitBookOrdermessage, which is as follows:/tns:submitBookOrder/tns:token
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….
This will launch the Create Correlation Set window. Give the correlation set a meaningful name, for example,
tokenCS.Next, click on the plus icon to launch the Property Chooser window, and select the
tokenproperty. - Next, 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. This will add an empty correlation to the Receive activity.
For the Correlation Set field, select tokenCS from the dropdown. Next, select Yes from the Initiate dropdown.
- 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 tokenCS from the dropdown. Next, select No from the Initiate dropdown.
- Deploy the
PublisherSkynetcomposite to the Oracle SOA Suite and use Enterprise Manager to submit multiplesubmitBookOrdermessages.For book orders that are submitted in the same minute, you should see that they are routed through to the same instance of the BPEL process.
This recipe makes use of the message aggregation feature we introduced in the first recipe to route multiple messages to the same process instance so that we can process them sequentially, thus ensuring that only a single request is submitted at a time to our external resource.
As before, the first message is routed to create a new instance of our singleton; subsequent messages then need to be routed through to this in-flight instance that is achieved by the use of a shared token , which is used to correlate messages against the singleton instance.
With a typical implementation of a singleton, we would create a single instance of that object and it would live forever. However, with BPEL, we can't let the process loop forever as, over time, the size of the audit trail for the BPEL instance within the dehydration store would get too big and impact the overall performance of the BPEL engine.
Thus, in reality, we need an instance of the singleton process to run for a period of time, and then, terminate gracefully at the end of this period, to be replaced by a new instance of the BPEL process.
For the purpose of this recipe, we have assumed that an instance of the singleton will run for just 1 minute (obviously, in reality we would allow a longer duration, for example, 24 hours, but it would take a long time to test the recipe).
So, for this purpose, we are using the current time (based on a shared system clock) to create the token. Take a look at the following statement:
substring( xp20:current-dateTime(), 1, 16)
The preceding statement creates a token of the format CCYY-MM-DDThh:mm. Thus, every message received in the same minute will have the same token and get routed through to the same instance of the BPEL process.
The process itself will only wait until CCYY-MM-DDThh:mm + 1 minute for new messages before terminating.
It should be noted that this approach isn't perfect. An obvious issue is what happens if the singleton is sent more messages than it is able to process within its execution window.
In this scenario, the messages would be queued up for delivery, but because the singleton terminates at the end of its window, those messages would be left on the queue and would never get processed.
One answer here is for the singleton. Once it has completed its execution period to enter a second loop that retrieves each remaining message and forwards it to the singleton (via the wrapper process — PublisherSkynet in the previous example). This will generate a new token based on the new time and the message will get invoked by the new singleton.
Of course, with this scenario, there is still the possibility that a message could keep getting forwarded forever, but this would imply the throughput of messages requiring processing is greater than the external resource/system can process over a sustained period of time, and therefore, we are "resource constrained" by the backend system.
The other potential issue is that we could have a race condition between the end of one singleton process and the start of the next, with both processes trying to access the same resource at the same time. To reduce the likelihood of this, we could include a wait period at the start of the process, which comfortably allows sufficient time for the previous process to complete.