Configuring WSO2 ESB with Oracle™AQ as Messaging Media

"OracleAQ provides database integrated messaging functionality. Its' underlying system works with oracle database. OracleJMS (OJMS) is the JMS interface to the Advanced Queuing (AQ) feature in the Oracle database. OJMS has JNDI support, which could be used to get 'ConnectionFactory' at run time and the connections returned from it will be able to transparently connect to the JMS server."

I have written a detail tutorial about the configurations needed to connect AQ , which is published [1] at wso2 site..Thanks Edwin  for the great help on AQ and LDAP configurations..
[1]http://wso2.org/library/tutorials/2011/11/configuring-wso2-esb-with-oracle-as-messaging-media

Java JNDI based client for OJMS

To work with Java Naming and Directory Interface (JNDI) in OJMS , which is the JMS interface for AQ, we need to register the oracle database with LDAP server. JMS administrator can register ConnectionFactory objects in a LDAP server.
Lets check following sample code, which is used to register the connection factory objects @ LDAP server.

void register_Factory_in_LDAP() throws Exception {
        Hashtable env = new Hashtable();
        // ldap settings
        env.put(Context.INITIAL_CONTEXT_FACTORY, AQjmsConstants.INIT_CTX_FACTORY);
        env.put(Context.PROVIDER_URL, "ldap://localhost:10389/");
        env.put(AQjmsConstants.SERVER_DN, "cn=ORCL,cn=OracleContext,ou=Services, o=sgi,c=us");
        env.put(Context.SECURITY_PRINCIPAL, "uid=ratha,ou=Services, o=sgi,c=us");
        env.put(Context.SECURITY_CREDENTIALS, "secret");
        env.put(Context.SECURITY_AUTHENTICATION, "simple");

        String url = "jdbc:oracle:thin:@localhost:1521/orcl";
        Properties properties = new Properties();
        properties.setProperty("user", "ratha");
        properties.setProperty("password", "ratha");
        try {
            AQjmsFactory.registerConnectionFactory(env, "test_queue_factory", url, properties,
                                                   "queue");
            System.out.println("Connection factory craeted ");

        } catch (Exception e) {
            e.printStackTrace();
          }
    }

// DO lookup

 void get_Factory_from_LDAP() throws Exception {

        Hashtable env = new Hashtable();
        env.put(Context.INITIAL_CONTEXT_FACTORY, AQjmsConstants.INIT_CTX_FACTORY);
        // ldapserver is your LDAP host and 389 is your port
        env.put(Context.INITIAL_CONTEXT_FACTORY, AQjmsConstants.INIT_CTX_FACTORY);
        env.put(Context.PROVIDER_URL, "ldap://localhost:10388/");
        env.put(Context.SECURITY_PRINCIPAL, "uid=ratha,ou=Services, o=sgi,c=us");
        env.put(Context.SECURITY_CREDENTIALS, "secret");

        DirContext inictx = new InitialDirContext(env);
        inictx = (DirContext) inictx.lookup("cn=ORCL,cn=OracleContext,ou=Services, o=sgi,c=us");
        // go to the connection factory holder cn=OraclDBConnections
        DirContext connctx = (DirContext) inictx.lookup("cn=oracledbconnections");

        // get connection factory "test_queue_factory"
        QueueConnectionFactory qc_fact = (QueueConnectionFactory) connctx.lookup("cn=ratha");

        System.out.println("Factory look up success " + qc_fact.toString());
        QueueConnection QCon = qc_fact.createQueueConnection();
        System.out.println("Connection created " + QCon.toString());
       DirContext destctxQF = (DirContext) inictx.lookup("cn=OracleDBQueues");
        System.out.println("OracleDBQueues look up success " + destctxQF.toString());
        Queue queue = (Queue) destctxQF.lookup("cn=ratha.test");
        System.out.println("Queue look up success :" + queue.toString());

        Session session = QCon.createQueueSession(true, QueueSession.AUTO_ACKNOWLEDGE);
        QCon.start();
        QueueSender sender = ((QueueSession) session).createSender(queue);
        System.out.println("Sender creation success :" + sender.toString());
    
       String msg = "test";
       TextMessage message = session.createTextMessage(msg);
       sender.send(message);
       // MessageConsumer consumer = session.createConsumer(queue);
       // TextMessage msg2 = (TextMessage) consumer.receive();
       // System.out.println("MESSAGE RECEIVED " + msg2.getText());
}

Creating a queuetable @ OracleAQ gives "ORA-01017: invalid username/password" error

If you face such issue[1], when you try to create a 'queuetable' at OracleAQ (Even though you provided all authorizations and authentications correctly), check the following global parameter is set to 'false"

connect / as sysdba

# sqlplus sys/admin@orcl as sysdba;
# ALTER SYSTEM SET GLOBAL_TOPIC_ENABLED = FALSE;

[1]Error

oracle.jms.AQjmsException: ORA-01017: invalid username/password;
logon denied ORA-06512: at "SYS.DBMS_AQADM", line 81
at oracle.jms.AQjmsSession.createQueueTable(AQjmsSession.java:4803)
at oracle.jms.AQjmsSession.createQueueTable(AQjmsSession.java:4778)

Creating a queue in OracleAQ

Oracle provides database integrated messaging functionality,which is called OracleAQ. 

Oracle JMS (OJMS) is the JMS interface to the Oracle Database Streams Advanced Queuing (AQ) feature.

To work with OracleAQ, we need to install Oracle database server.(V11g)


In this post lets have a look on how we can create a queue programmatically using OJMS libraries and AQ libraries.
For this, specific user should have required authorization to create queues..

• Connect as 'sys/admin' ;
• create user ratha identified by ratha;
• grant create session to ratha;
• grant connect, resource to ratha;
  
• grant aq_administrator_role to ratha  identified by  ratha;
• grant execute on dbms_aq to  ratha;
• grant execute on dbms_aqadm to  ratha;
• exec dbms_aqadm.grant_system_privilege('ENQUEUE_ANY','ratha');
• exec dbms_aqadm.grant_system_privilege('DEQUEUE_ANY','ratha'); 
• grant execute on sys.aq$_jms_text_message to ratha; 
• Connect as 'ratha/ratha'

After providing required authorization for the new user 'ratha' (password : ratha) you could be able to login as user 'ratha' and able to create table/queue etc..

With the particular user information, now lets try to create a queue in oracle database.

import javax.jms.JMSException;
import javax.jms.MessageConsumer;
import javax.jms.MessageProducer;
import javax.jms.Queue;
import javax.jms.QueueConnection;
import javax.jms.QueueConnectionFactory;
import javax.jms.QueueSession;
import javax.jms.Session;
import javax.jms.TextMessage;
import oracle.AQ.AQQueueTable;
import oracle.AQ.AQQueueTableProperty;
import oracle.jms.AQjmsDestination;
import oracle.jms.AQjmsDestinationProperty;
import oracle.jms.AQjmsFactory;
import oracle.jms.AQjmsSession;

public class OracleAQClient {

public static QueueConnection getConnection() {

  String hostname = "localhost";
  String oracle_sid = "orcl";
  int portno = 1521;
  String userName = "ratha";
  String password = "ratha";
  String driver = "thin";
  QueueConnectionFactory QFac = null;
  QueueConnection QCon = null;
  try {
   // get connection factory , not going through JNDI here
   QFac = AQjmsFactory.getQueueConnectionFactory(hostname, oracle_sid, 
portno,driver);
   // create connection
   QCon = QFac.createQueueConnection(userName, password);
   } catch (Exception e) {
   e.printStackTrace();
  }
  return QCon;
 }

 public static void createQueue(Session session, String user,
String qTable, String queueName) {
  try {
   /* Create Queue Tables */
   System.out.println("Creating Queue Table...");

   AQQueueTableProperty qt_prop;
   AQQueueTable q_table = null;
   AQjmsDestinationProperty dest_prop;
   Queue queue = null;
   qt_prop = new AQQueueTableProperty("SYS.AQ$_JMS_TEXT_MESSAGE");

   /* create a queue table *///
   // /* Drop the queue table if already exists */
   // try{
   // q_table = ((AQjmsSession) session).getQueueTable(user, qTable);
   // q_table.drop(true);
   // System.out.println("Droped older queuetable...");
   // }
   // catch(Exception e){
   // e.printStackTrace();
   // return;
   // }

   q_table = ((AQjmsSession) session).createQueueTable(user, qTable,
qt_prop);
   System.out.println("Qtable created");
   dest_prop = new AQjmsDestinationProperty();
   /* create a queue */
   queue = ((AQjmsSession) session).createQueue(q_table, queueName, 
dest_prop);
   System.out.println("Queue created");
   /* start the queue */
   ((AQjmsDestination) queue).start(session, true, true);
  } catch (Exception e) {
   e.printStackTrace();
   return;
  }
 }

 public static void sendMessage(String user, String queueName) {

  try {
   QueueConnection QCon = getConnection();  
   Session session = QCon.createQueueSession(false,
Session.CLIENT_ACKNOWLEDGE);
   QCon.start();
   Queue queue = ((AQjmsSession) session).getQueue(user, queueName);
   MessageProducer producer = session.createProducer(queue);
   TextMessage tMsg = session.createTextMessage("test");
   producer.send(tMsg);
   System.out.println("Sent message = " + tMsg.getText());

   session.close();
   producer.close();
   QCon.close();

  } catch (JMSException e) {
   e.printStackTrace();
   return;
  }
 }

 public static void browseMessage(String user, String queueName) {
  Queue queue;
  try {
   QueueConnection QCon = getConnection();  
   Session session = QCon.createQueueSession(false,
Session.CLIENT_ACKNOWLEDGE);
  
   QCon.start();
   queue = ((AQjmsSession) session).getQueue(user, queueName);
   QueueBrowser browser = session.createBrowser(queue);
   Enumeration enu = browser.getEnumeration();
   List list = new ArrayList();  
   while (enu.hasMoreElements()) {
    TextMessage message = (TextMessage) enu.nextElement();   
    list.add(message.getText());
   }
   for (int i = 0; i < list.size(); i++) {
    System.out.println("Browsed msg " + list.get(i));
   }
   browser.close();
   session.close();
   QCon.close();

  } catch (JMSException e) {
   e.printStackTrace();
  }

 }

 public static void consumeMessage(String user, String queueName) {  
  Queue queue;
  try {
   QueueConnection QCon = getConnection();  
   Session session = QCon.createQueueSession(false,
Session.CLIENT_ACKNOWLEDGE);
   QCon.start();
   queue = ((AQjmsSession) session).getQueue(user, queueName);
   MessageConsumer consumer = session.createConsumer(queue);
   TextMessage msg = (TextMessage) consumer.receive();
   System.out.println("MESSAGE RECEIVED " + msg.getText());

   consumer.close();
   session.close();
   QCon.close();
  } catch (JMSException e) {  
   e.printStackTrace();
  }
 }

 public static void main(String args[]) {
  String userName = "ratha";
  String queue = "test";
  // createQueue( userName, qTable, queue);
  sendMessage(userName, queue);
  browseMessage(userName, queue);
  // consumeMessage(userName, queue);
 }
}

You might need following jars in your class-path in order to run above java client

• ojdbc6.jar (can be found at db_home\jdbc\lib)
• jta.jar (can be found at db_home\jdbc\jlib)
• jmscommon.jar (can be found at db_home\RDBMS\jlib folder)
• aqapi.jar(can be found at db_home\RDBMS\jlib folder)