Working with AQ Adapter in SOA

Introduction to the Oracle AQ Adapter:

Oracle Streams Advanced Queuing (AQ) provides a flexible mechanism for bidirectional, asynchronous communication between participating applications. Advanced queues are an Oracle database feature, and are therefore scalable and reliable. Other features of Oracle database, such as backup and recovery (including any-point-in-time recovery), logging, transactional services, and system management, are also inherited by advanced queues. Multiple queues can also service a single application, partitioning messages in a variety of ways and providing another level of scalability through load balancing.

Oracle AQ Adapter Features:

• The Oracle AQ Adapter is both a producer and a consumer of AQ messages. The enqueue operation is exposed as a JCA outbound interaction. The dequeue operation is exposed as a JCA inbound interaction.
• The Oracle AQ Adapter supports ADT (Oracle object type), XMLType, and RAW queues as payloads. It also supports extracting a payload from one ADT member column.
• The Oracle AQ Adapter supports normalized properties for enqueue and dequeue operations.

·         The Oracle AQ Adapter supports the following features of Oracle Streams AQ:

o    Correlation Identifier

o    Multiconsumer Queue

o    Message Priority

o    Time Specification and Scheduling and many more.

(For more about AQ Adapter refer Oracle® Fusion Middleware User's Guide for Technology Adapters 11g Release 1 (11.1.1.6.0)).

SQL Script to create the AQ table & Message Type:

SQL>CONNECT sys/change_on_install as sysdba

SQL>DROP USER aq_user CASCADE;

SQL>CREATE USER aq_user IDENTIFIED BY aq_user

  DEFAULT TABLESPACE users

  TEMPORARY TABLESPACE temp;

SQL>ALTER USER aq_user QUOTA UNLIMITED ON users;

SQL>GRANT aq_administrator_role TO aq_user;

SQL>GRANT connect               TO aq_user;

SQL>GRANT create type           TO aq_user;

SQL>GRANT create sequence       TO aq_user;

SQL>EXECUTE dbms_aqadm.grant_type_access ('aq_user');

-- Login as AQ_User

SQL>CONNECT aq_user/aq_user

SQL>CREATE TYPE message_type AS OBJECT (message_id NUMBER (15),subject VARCHAR2(100),text VARCHAR2(100),dollarvalue NUMBER(4,2));

SQL>BEGIN

   -- ----------------------------------------------------

    DBMS_AQADM.CREATE_QUEUE_TABLE (

        queue_table        => 'aq_user.msg_qt'

      , queue_payload_type => 'aq_user.message_type'

    );

    -- ----------------------------------------------------

    DBMS_AQADM.CREATE_QUEUE (

        queue_name          => 'msg_queue'

      , queue_table         => 'aq_user.msg_qt'

      , queue_type          => DBMS_AQADM.NORMAL_QUEUE

      , max_retries         => 0

      , retry_delay         => 0

      , retention_time      => 1209600

      , dependency_tracking => FALSE

      , comment             => 'Test Object Type Queue'

      , auto_commit         => FALSE

    );

    -- ----------------------------------------------------

    DBMS_AQADM.START_QUEUE ('msg_queue');

    -- ----------------------------------------------------

END;

Script to Stop and Drop the Queue:

SQL>CONNECT aq_user/aq_user

EXECUTE dbms_aqadm.stop_queue (queue_name => 'aq_user.msg_queue');

EXECUTE dbms_aqadm.drop_queue (queue_name => 'aq_user.msg_queue');

EXECUTE

dbms_aqadm.drop_queue_table (queue_table => 'aq_user.msg_qt');

SQL>DROP TYPE aq_user.message_type;

 


Weblogic configuration for AQ Adapter:

Login to
Weblogic server console. Then create a Data source. For this go to the Data
Sources section. Create JDBC Data source localhost-aq whose JNDI name is jndi/localhost-aq
 

Now under Deployments select AQAdapter, Configurations Tab and Outbound Connection, Create a new Outbound Connection eis/aq/localhost-aq who’s XADataSourceName is jndi/localhost-aq

Below are Data source details to aq_user user in database

This will be deployed on Admin Server, as my SOA installation is running on Admin server.

Save

 

Update the AQAdapter Deployment, Update the plan, no restart needed.

Soa 11g bpel process using enqueue:

Let’s create a simple Asynchronous BPEL process my-aq-app, Under External Reference Insert AQ Adapter.

Edit the BPEL Process use Invoke activity to connect to AQ Adapter 

 

This is how the BPEL would look like 

 

Add assign before invoke to wire input variables to AQ input variables.

  

These are Details of AQ Adapter

Create a DB Connection by name localhost-aq

Note: JNDI name should match to the one that we configured in the Weblogic Console