You may want to process inbound events sequentially. The problem is that the adapter raises events/records asynchronously to BPEL, thatt are processed by multiple threads. So it is possible that a record raised 2nd by the adapter may be processed first in BPEL.
To alter this you can make the inbound polling adapter to BPEL exchange synchronous, meaning thread control will not pass back to the adapter until the record has been processed in BPEL. This means that the inbound adapter can only raise one record/event at a time, erasing performance gains from multi-threading but guaranteeing in-order processing in BPEL.
Following is an example of configuring a DBAdapter wsdl to do a synchronous exchange. The idea is to add an empty reply to the adapter-BPEL exchange, making the inbound adapter thread wait for this reply until BPEL processing has completed. For the complete instructions look at bpel/samples/tutorials/122.DBAdapter/advanced/endToEnd/DirectSQLPerformance/README.txt
1. First, open up the receiveService.wsdl, and modify the receive operation to
have both an input and an output.
have both an input and an output.
I.e. change:
<portType name="receiveService_ptt">
<operation name="receive">
<input message="tns:PerfInCollection_msg"/>
</operation>
</portType>
<operation name="receive">
<input message="tns:PerfInCollection_msg"/>
</operation>
</portType>
to:
<portType name="receiveService_ptt">
<operation name="receive">
<input message="tns:PerfInCollection_msg"/>
<output message="tns:PerfOut_msg"/>
</operation>
</portType>
<operation name="receive">
<input message="tns:PerfInCollection_msg"/>
<output message="tns:PerfOut_msg"/>
</operation>
</portType>
2. Then you will need define this 'PerfOut_msg', copy and paste the input
message and rename it, so you go from having i.e.:
message and rename it, so you go from having i.e.:
<message name="PerfInCollection_msg">
<part name="PerfInCollection" element="top:PerfInCollection"/>
</message>
<part name="PerfInCollection" element="top:PerfInCollection"/>
</message>
to:
<message name="PerfInCollection_msg">
<part name="PerfInCollection" element="top:PerfInCollection"/>
</message>
<part name="PerfInCollection" element="top:PerfInCollection"/>
</message>
<message name="PerfOut_msg">
<part name="PerfOut" element="top:empty"/>
</message>
<part name="PerfOut" element="top:empty"/>
</message>
3. The binding section also needs to have an extra <output/> tag added.
Replace i.e.
<binding name="receiveService_binding" type="tns:receiveService_ptt">
<pc:inbound_binding/>
<operation name="receive">
<jca:operation
ActivationSpec="oracle.tip.adapter.db.DBActivationSpec"
<pc:inbound_binding/>
<operation name="receive">
<jca:operation
ActivationSpec="oracle.tip.adapter.db.DBActivationSpec"
DescriptorName="DirectSQLPerformance.PerfIn"
QueryName="receiveService"
PollingStrategyName="DeletePollingStrategy"
MaxRaiseSize="100"
MaxTransactionSize="100"
PollingInterval="5"
NumberOfThreads="10"
MappingsMetaDataURL="toplink_mappings.xml" />
<input/>
</operation>
</binding>
QueryName="receiveService"
PollingStrategyName="DeletePollingStrategy"
MaxRaiseSize="100"
MaxTransactionSize="100"
PollingInterval="5"
NumberOfThreads="10"
MappingsMetaDataURL="toplink_mappings.xml" />
<input/>
</operation>
</binding>
with:
<binding name="receiveService_binding" type="tns:receiveService_ptt">
<pc:inbound_binding/>
<operation name="receive">
<jca:operation
ActivationSpec="oracle.tip.adapter.db.DBActivationSpec"
<pc:inbound_binding/>
<operation name="receive">
<jca:operation
ActivationSpec="oracle.tip.adapter.db.DBActivationSpec"
DescriptorName="DirectSQLPerformance.PerfIn"
QueryName="receiveService"
PollingStrategyName="DeletePollingStrategy"
MaxRaiseSize="100"
MaxTransactionSize="100"
PollingInterval="5"
NumberOfThreads="10"
MappingsMetaDataURL="toplink_mappings.xml" />
<input/>
<output/>
</operation>
</binding>
QueryName="receiveService"
PollingStrategyName="DeletePollingStrategy"
MaxRaiseSize="100"
MaxTransactionSize="100"
PollingInterval="5"
NumberOfThreads="10"
MappingsMetaDataURL="toplink_mappings.xml" />
<input/>
<output/>
</operation>
</binding>
4. Now open up the corresponding xsd. In step 2 you indicated that the return
message was going to be of xml type "top:empty" where xml elements are defined
in your xsd. Simply open it up and copy and paste the below into the bottom
of the file.
message was going to be of xml type "top:empty" where xml elements are defined
in your xsd. Simply open it up and copy and paste the below into the bottom
of the file.
<xs:element name="empty" type="empty"/>
<xs:complexType name="empty">
<xs:sequence/>
</xs:complexType>
<xs:complexType name="empty">
<xs:sequence/>
</xs:complexType>
I.e. so it looks like
<xs:element name="empty" type="empty"/>
<xs:complexType name="empty">
<xs:sequence/>
</xs:complexType>
</xs:schema>
<xs:complexType name="empty">
<xs:sequence/>
</xs:complexType>
</xs:schema>
5. Now in your bpel process you need to create a 'reply' activity at the end
of the BPEL process. Point the 'reply' to the polling parternlink and auto-
create the input variable (which is just empty). Normally you only create
a 'receive' for the dbadapter polling partnerlink.
of the BPEL process. Point the 'reply' to the polling parternlink and auto-
create the input variable (which is just empty). Normally you only create
a 'receive' for the dbadapter polling partnerlink.
Make sure the BPEL process itself has no dehydration points,
0 comments:
Post a Comment