Using Oracle AQ with Oracle SOA 11g & OSB 11g

Oracle Advanced Queuing(AQ) is Oracle Database message queuing mechanism which acts as the queuing infrastructure.  With AQ the different applications can communicate and exchange the message via Oracle database queues.  Oracle AQ provides the API to enqueue and dequeue the messages to and from the queues in Oracle database.

Oracel AQ is the integral part of Oracle database already.  In this tutorial Oracle XE 10g is used.   Oracle AQ is available from two built-in PL/SQL packages in Oracle database: DBMS_AQADM and DBMS_AQ.

DBMS_AQADM package provides PL/SQL procedures to manage the AQ configuration and administration.  Some basic functionality in DBMS_AQADM are like: Create the queue table, create the queue, start the queue, stop queue.

DBMS_AQ package provides the interface to other applications to AQ.  DEQUEUE and ENQUEUE are two procedures for enqeuing and deqeuing the AQ queue.

After one AQ queue is created and started the application such as PL/SQL procedures can enqueue the message  into the queue and other application such as one SOA component can dequeue the message from this queue.

The below are the steps showing how one AQ queue is created from the scratch and how a message is enqueued in the queue and etc.  After these the steps of creating one SOA component to dequeue the message from AQ queue.

Setup AQ queue in Oracle XE 10g

There are several steps in setting up one AQ queue in Oracle database:

1.     Create one user in database.  This user will be granted with the roles: AQ_ADMINISTRATOR_ROLE, AQ_USER_ROLE.  Log into Oracle XE 10g as the sysdba and then create one user called EVENT_USER and then grant certain privileges to this user.

sqlplus sys@localhost:1521/xe as sysdba
create user EVENT_USER identified by oracle;
grant create session, resource, AQ_ADMINISTRATOR_ROLE,AQ_USER_ROLE to EVENT_USER;
grant execute on DBMS_AQADM to EVENT_USER;
grant execute on DBMS_AQ to EVENT_USER;

2.     Create a queue table.  Before one type called invoice_type is created.  This type is used as the payload type of the message in AQ queue.   

create or replace type invoice_type as object(
      invoice_id number,
      invoice_number varchar2(60),
      vendor_id number
);
/

exec DBMS_AQADM.CREATE_QUEUE_TABLE ( 
              queue_table => 'EVENT_USER.invoice_queue_table', 
              queue_payload_type => 'EVENT_USER.invoice_type');

SELECT queue_table,type,object_type,recipients FROM USER_QUEUE_TABLES;

3.     After the queue table is created create a queue

exec DBMS_AQADM.CREATE_QUEUE (
          queue_name =>'EVENT_USER.invoice_queue',              
          queue_table=>'EVENT_USER.invoice_queue_table'); 

select name, queue_table, queue_type from user_queues;

After the execution of above PL/SQL scripts some database objects are created for this AQ queue as shown below. 

4.     Last step is to start the queue.  It can be done by executing the below procedure from DBMS_AQADM.

exec DBMS_AQADM.START_QUEUE('EVENT_USER.invoice_queue');

Enqueue AQ queue with PL/SQL package

Now the queue is ready to enqueue a message in the AQ queue just created and started. In this example one PL/SQL procedure from DBMS_AQ package is used to enqueue the AQ queue.

DECLARE
  enqueue_options    dbms_aq.enqueue_options_t;
  message_properties       dbms_aq.message_properties_t;
  message_handle     RAW(16);
  message           invoice_type ;
  message_id         NUMBER;
BEGIN

 message := invoice_type (10, 'ABCD123', 80);
 enqueue_options.VISIBILITY := DBMS_AQ.ON_COMMIT;
  enqueue_options.SEQUENCE_DEVIATION := null;
  message_properties.EXPIRATION := DBMS_AQ.NEVER;

    DBMS_AQ.ENQUEUE (
    queue_name => 'invoice_queue',
    enqueue_options => enqueue_options,
    message_properties => message_properties,
    payload => message,
    msgid => message_handle);
    
  COMMIT;
END;
/

The message just enqueued can be browsed using the below query:

SELECT count(*) FROM aq$invoice_queue_table

SELECT user_data FROM   aq$invoice_queue_table;

Consume the message in AQ queue from Oracle SOA component

Oracle SOA component(BPEL and Mediator) can  consume the message from AQ queue(dequeue) or produce the message into the queue(enqueue). It is done by using Oracle SOA uses JCA adapter for AQ.

In order to use AQ adapter some configurations need to be done.

1.     Create the data source in Weblogic console via Domain Structure->Services->Data Sources.  

This data source will point to the Oracle database where AQ queue resides on.  In this example it is Oracle XE 10g running on local machine. Here AQEvent data source is created and its JNDI name is set as jdbc/AQEvent. 

2.     Create AQAdpater instance in Weblogic console via Domain Structure->Deployments->Deployments->AQAdapter.  

An new AQ adapter instance is created as eis/AQ/LocalEventConnection whose XADataSourceName is set as: jdbc/AQEvent.

3.     Use AQAdapter in JDeveloper

Here the AQAdapter will be used as the service to consume the message from the AQ queue specified in adapter jca file.  This jca file also specifies the AQ connection factory which is configured as eis/AQ/LocalEventConnection in the above step.

<adapter-config adapter="AQ Adapter" name="AQEventService" wsdllocation="AQEventService.wsdl" xmlns="http://platform.integration.oracle/blocks/adapter/fw/metadata">
  
  <connection-factory location="eis/AQ/LocalEventConnectionen" uiconnectionname="LocalEventConnectionen">
  <endpoint-activation operation="Dequeue" porttype="Dequeue_ptt">
    <activation-spec classname="oracle.tip.adapter.aq.inbound.AQDequeueActivationSpec">
      <property name="SchemaValidation" value="false">
      <property name="QueueName" value="INVOICE_QUEUE">
      <property name="DatabaseSchema" value="EVENT_USER">
    </property></property></property></activation-spec>
  </endpoint-activation>

</connection-factory>
</adapter-config>

After the SOA project is deployed to the server, the AQAdapter will start to poll the message in AQInvoice queue.

Consume the message in AQ queue from OSB
In OSB JCA adapter is also used to poll the message in AQ queue.  Here the AQ JCA used in SOA will be used to generate one proxy service.   Also assume that the same data source and AQAdapter instance are configured in OSB server.

Create one OSB project as AQTestProject.
Copy the below files from JDeveloper to the OSB project:
                  AQEventService_aq.jca
                  AQEventService.wsdl
                  xsd/EVENT_USER_INVOICE_TYPE.xsd

From .jca file to generate the OSB proxy service.

The generated proxy service will be as the below:

Design the message flow as below:

After the OSB project is deployed on the server it will start to consume the message in AQ.

DbAdapter instance is not setup correctly in cluster environment

Recently I had the issue one Oracle SOA DbAdapter instance in the cluster environment.   The DbAdapter is created in the single node development environment.  Everything is working fine but after moving to testing environment which is cluster environment.   DbAdapter is created following the same steps as did for development environment.  Things start to go wrong sometimes.  The faults occurred complains that
the Resource Adapters RAR file has not been deployed successfully to the WebLogic Application server or
element in weblogic-ra.xml has not been set to  the created DbApater instance.  However the DbApater instance looks completely correct from Weblogic Console.   After some while of investigation it is realized that it might be the resource adapter plan file is not set up correctly in both nodes or one of node.  Checking the Plan.xml files on each node verifies this.   After copying the Plan which is set correctly to another node everything starts to work fine.

There is one wonderful blog which explains this very clearly.
http://myexperienceswithsoa.blogspot.com.au/2012/02/planxml-not-getting-updated-across.html

So next time when you are working with resource adapter in cluster environment keep this in mind. 

Use logical directory path in File Adapter

Oracle SOA File adapter provides both physical and logical directory path in configuring File adapter. By its name the physical path is the physical path of the server. Obviously it is not so flexible to use physical path in the design time. Logical path provides advantages over the physical path. Firstly you can deploy the same composite into the various environments where the actual physical path may be different. For example in development environment the path is /u90/uminput while in the production the path is /u100/shared/uminput. Also with logical path you can change the physical path during the runtime without redeployment. 

Set logical path in FILE Adapter Configuration Wizard 

The first step is to select to use logical path when configuring the file adapter as shown in the below. Here XML_FILE_PATH is the logical name of the directory path for the File adapter.

composite.xml 

After the configuration the file adapter using File Adapter Wizard there will one property called XML_FILE_PATH for the service in composite.xml.

cfgplan.xml 

For each environment the different value can be set for the property XML_FILE_PATH in the configuration plan. The value is the actual physical path in the environment.

Enterprise Manager Console 

Also the value for XML_FILE_PATH can be set at runtime by using Enterprise Manager Console as shown in below.

File Adapter in Oracle SOA 11g cluster environment

A couple days ago I had an issue with Oracle SOA Adapter in cluster environment.  File Adapter is used as inbound file reader in the composite. It works fine with the development environment which is single node.  The issue is that the file will be read twice, each time by a File Adapter instance on one cluster node.

From the search for the solution of this issue there are two ways to tackle this problem:

1. Use Singleton property for the inbound endppoint for SOA composite.
To enable this just add singletom property to composite.xml.

 
 <service name="InvoiceXMLReader" ui:wsdlLocation="InvoiceXMLReader.wsdl">

    <interface.wsdl interface="http://xmlns.oracle.com/pcbpel/adapter/file/EInvoiceProcessing/InvoiceXMLReader/InvoiceXMLReader#wsdl.interface(Read_ptt)"/>

    <binding.jca config="InvoiceXMLReader_file.jca">

        <property name="singleton">true</property>

    </binding.jca>

  </service>

2. To use HAFielAdapter for File Adapter.

Thus in the .jca fiel change to use eis/HAFileAdapter instead of eis/FileAdapter.

 
<adapter-config name="InvoiceXMLReader" adapter="File Adapter" wsdlLocation="InvoiceXMLReader.wsdl" xmlns="http://platform.integration.oracle/blocks/adapter/fw/metadata">

 

  <connection-factory UIexcludeWildcard="*.tif;*.lock" location="eis/HAFileAdapter" UIincludeWildcard="*.*"/>

  <endpoint-activation portType="Read_ptt" operation="Read">

    <activation-spec className="oracle.tip.adapter.file.inbound.FileActivationSpec">

      <property name="DeleteFile" value="false"/>

      <property name="MinimumAge" value="0"/>

      <property name="PhysicalDirectory" value="/u01/shared/einvoice"/>

      <property name="Recursive" value="true"/>

      <property name="PollingFrequency" value="10"/>

      <property name="IncludeFiles" value=".*\..*"/>

      <property name="UseHeaders" value="false"/>

      <property name="ExcludeFiles" value=".*\.tif;.*\.lock"/>

    </activation-spec>

  </endpoint-activation>

</adapter-config>

Behind HAFileAdapter is the database used as mutex to ensure the one file is only handled by one instance.  By default it uses jdbc/SOADataSource.   When using HAFileAdapter in OSB this jdbc/SOADataSource should also target the OSB server nodes as well.
This also invovles some configuration changes for File Adapter.  You need to set Deployments->File Adapter->Configuration->Outbound Connection Pools->javax.resource.cci.ConnectionFactory->eisHAFileAdapter->controlDir to some shared localtion.

To my understanding the first way is to force to seralize the inbound file processing, which means only one File Adapter is processing a file at one time.   HAFileAdapter is using some mutex to ensure one file is only processed by one FileAdapter instance.  But the multiple HAFileAdapter can process the different file at the same time.  So this is the real solution for high availability environment.