/* $Header: aqjmsdemo06.java 13-jul-2000.11:22:23 rbhyrava Exp $ */

/* Copyright (c) Oracle Corporation 2000. All Rights Reserved. */

/*
   DESCRIPTION
    <short description of component this file declares/defines>

   PRIVATE CLASSES
    <list of private classes defined - with one-line descriptions>

   NOTES
    <other useful comments, qualifications, etc.>

   MODIFIED    (MM/DD/YY)
    rbhyrava    07/13/00 - fix typo
    rbhyrava    03/16/00 - AQ jms demos
    rbhyrava    03/15/00 - AQ JMS demo - Propagation
    rbhyrava    03/15/00 - Creation
 */

/**
 *  @version $Header: aqjmsdemo06.java 13-jul-2000.11:22:23 rbhyrava Exp $
 *  @author  rbhyrava
 *  @since   release specific (what release of product did this appear in)
 */

/***
 * This is a sample java file which uses Oracle JMS - Java Messaging Service 
 * API to schedule Propagation between Queues in Oracle database 
 *
 * This demo is functionally same a newaqdemo01.sql and does the following:
 * -- Create two queue tables - input_queue_table, prop_queue_table
 * -- Create two queues - input_queue belonging to input_queue_table,
 *    prop_queue belonging to prop_queue_table
 * -- Create two subscribers to input_queue - prog1, prog2
 * -- Create one subscribers to input_queue - prog3 at prop_queue
 * -- Schedule propagation between input_queue and other queues in
 *    the database
 *
 * The following instructions describe how to compile and execute 
 * this sample on the client machine. 
 *
 * System requirements:
 * ====================
 *  1) Oracle 8.1.6 database or higher 
 *  2) The client machine should have JDK 1.1.x or JDK1.2 or higher installed
 *  3) The following jar/zip files should be in the CLASSPATH on the client 
 *     machine. 
 *     For JDK1.2.x 
 *        classes12.zip 
 *        aqapi.jar
 *        jmscommon.jar
 *     For JDK1.1.x 
 *        classes111.zip 
 *        aqapi11.jar
 *        jmscommon.jar
 *  Set up CLASSPATH, PATH, LD_LIBRARY_PATH based on JDK version and platform.
 * Compilation and Running:
 *  ========================
 *  4) If you already have the jars in step 3) in classpath 
 *      javac aqjmsdemo06.java Message.java 
 *
 *  5) java aqjmsdemo06  <SID> <HOST> <PORT> <DRIVER> 
 *     Example usage: 
 *       java aqjmsdemo06 orcl82 dlsun666 1521 thin 
 *
 ***/

/* Object Message */

/* import useful packages */
import oracle.AQ.*;
import oracle.jms.*;
import javax.jms.*;
import java.lang.*;

public class aqjmsdemo06
{
  public static void main (String args [])
       throws java.sql.SQLException, ClassNotFoundException, JMSException
  {
     TopicSession  tsess = null;
      try
      {
        if (args.length < 4 )
         System.out.println("Usage:java filename [SID] [HOST] [PORT] [DRIVER]");
        else {
          tsess = createSession(args);
          setupTopic(tsess) ;
          performJmsOperations(tsess);
          System.out.println("End of Demo") ;
        }
      }
      catch (Exception ex)
      {
         System.out.println("Exception-1: " + ex);
         ex.printStackTrace();
      }
  }

   public static TopicSession createSession(String args[]) 
     throws Exception
   {
      TopicSession  tsess = null;
      TopicConnectionFactory tcfact;
      TopicConnection tconn;
      try 
      {
        //args passed are sid,host,port,driver 
         
         tcfact = AQjmsFactory.getTopicConnectionFactory(
                      args[1], args[0], Integer.parseInt(args[2]), args[3]);

         tconn = tcfact.createTopicConnection( "jmsuser","jmsuser");

         /* Create a Topic Session */
         tsess = 
             tconn.createTopicSession(true, Session.CLIENT_ACKNOWLEDGE);

         tconn.start() ;
         System.out.println("Successfully created TopicSession");  
      }
      catch (Exception ex)
      {
         System.out.println("Exception: " + ex); 
         throw ex;
      }  
      return tsess;
   }

   public static void setupTopic(TopicSession tsess) throws Exception
   {
      AQQueueTableProperty qtprop1,qtprop2 ;
      AQQueueTable qtable,table1, table2;
      AQjmsDestinationProperty dprop;
      Topic topic1, topic2;
      try {
         /* Create Queue Tables */
         System.out.println("Creating Input Queue Table...") ;

         /* Drop the queue if already exists */
         try {
           qtable=((AQjmsSession)tsess).getQueueTable("jmsuser", 
                  "INPUT_QUEUE_TABLE" );
           qtable.drop(true);
         } catch (Exception e) {} ;
         try {
           qtable=((AQjmsSession)tsess).getQueueTable("jmsuser", 
                  "PROP_QUEUE_TABLE" );
           qtable.drop(true);
         } catch (Exception e) {} ;

         qtprop1 = new AQQueueTableProperty ("SYS.AQ$_JMS_OBJECT_MESSAGE") ;
         qtprop1.setComment("input queue") ;
         qtprop1.setMultiConsumer(true) ;
         qtprop1.setCompatible("8.1") ;
         qtprop1.setPayloadType("SYS.AQ$_JMS_OBJECT_MESSAGE") ;
         table1 = ((AQjmsSession)tsess).createQueueTable("JMSUSER", 
           "INPUT_QUEUE_TABLE", qtprop1) ;
  
         System.out.println("Creating Propagation Queue Table...") ;
         qtprop2 = new AQQueueTableProperty ("SYS.AQ$_JMS_OBJECT_MESSAGE") ;
         qtprop2.setComment("Popagation queue") ;
         qtprop2.setPayloadType("SYS.AQ$_JMS_OBJECT_MESSAGE") ;
         qtprop2.setMultiConsumer(true) ;
         qtprop2.setCompatible("8.1") ;
         table2 = ((AQjmsSession)tsess).createQueueTable("JMSUSER", 
           "PROP_QUEUE_TABLE", qtprop2) ;
  
        System.out.println ("Creating Topic input_queue...");
        dprop = new AQjmsDestinationProperty() ;
        dprop.setComment("create topic 1") ;
        topic1=((AQjmsSession)tsess).createTopic(table1,"INPUT_QUEUE",dprop) ;
  
        dprop.setComment("create topic 2") ;
        topic2=((AQjmsSession)tsess).createTopic( table2,"PROP_QUEUE",dprop) ;
  
        /* Start the topic */
        ((AQjmsDestination)topic1).start(tsess, true, true);
        ((AQjmsDestination)topic2).start(tsess, true, true);
        System.out.println("Successfully setup Topics");  

      } catch (Exception ex) {
         System.out.println("Error in setupTopic: " + ex);
         throw ex;
      }

   }


  public static void performJmsOperations(TopicSession tsess) 
  throws Exception
  {
    Topic topic1,topic2;
    TopicSubscriber[] subs;
    AQjmsAgent agt;
    ObjectMessage objmsg  = null, robjmsg=null;
    TopicPublisher publisher ;
    Message  sobj , rmsg;
    String[] cities={"BELMONT","REDWOOD SHORES", "SUNNYVALE", "BURLINGAME" };
    try
    {
      
      System.out.println("Get Topics...") ;
      topic1 = ((AQjmsSession)tsess).getTopic("JMSUSER", "INPUT_QUEUE") ;
      topic2 = ((AQjmsSession)tsess).getTopic("JMSUSER", "PROP_QUEUE") ;

      System.out.println("Creating Topic Subscribers...") ;

      subs = new TopicSubscriber[3] ;
      subs[0] = ((AQjmsSession)tsess).createDurableSubscriber(
                                  topic1, "PROG1", null, false);
      subs[1] = ((AQjmsSession)tsess).createDurableSubscriber(
                                  topic1, "PROG2", "JMSPriority > 2", false);
      subs[2] = tsess.createDurableSubscriber(
                                  topic2, "PROG3", null , false) ;

      agt = new AQjmsAgent("PROG3", "PROP_QUEUE" ) ;

      System.out.println("Creating Remote Subscriber...") ;
      ((AQjmsSession)tsess).createRemoteSubscriber(
                                  topic1, agt,"JMSPriority = 2");

      /* Schedule Propagation with latency 0 */ 
      System.out.println("Schedule Propagation...") ;
      ((AQjmsDestination)topic1).schedulePropagation(
                                tsess, null, null, null, null, new Double(0)) ;

      System.out.println("Publish messages...") ;
      objmsg = ((AQjmsSession)tsess).createObjectMessage() ;

      publisher = tsess.createPublisher(topic1); 

      /*publish 100 messages*/

      for ( int i = 1 ; i <= 100 ; i++) 
      {
         objmsg.setIntProperty("Id",i) ;
         if ( ( i % 3 ) == 0 )  { 
           objmsg.setStringProperty("City",cities[0]) ;
         } 
         else if ((i % 4 ) == 0) { 
           objmsg.setStringProperty("City",cities[1]) ;
         }
         else if (( i % 2) == 0) {
           objmsg.setStringProperty("City",cities[2]) ;
         }
         else {
           objmsg.setStringProperty("City",cities[3]) ;
         }

         objmsg.setIntProperty("Priority",(1+ (i%3))) ;

         sobj = new Message() ;
         sobj.setId(i) ;
         sobj.setName("message# "+i) ;
         sobj.setData(500);
         objmsg.setObject(sobj) ;
         objmsg.setJMSCorrelationID(""+i) ;
         objmsg.setJMSPriority(1+(i%3)) ;
         publisher.publish(topic1,objmsg, DeliveryMode.PERSISTENT, 
             1 +(i%3), AQjmsConstants.EXPIRATION_NEVER);
      }
      System.out.println("Commit now...") ;
      tsess.commit() ;

      Thread.sleep(50000);

      /* Receive messages for each subscriber */
      System.out.println("Receive Messages...") ;
      for (int i=0; i< subs.length ; i++) 
      {
         System.out.println ("Messages for subscriber : " +i) ;
         if (subs[i].getMessageSelector() != null)
         {
            System.out.println("  with selector: " +
                             subs[i].getMessageSelector());
         }

         boolean done = false;
         while(!done) {
         try {
           robjmsg = (ObjectMessage)( subs[i].receiveNoWait() );
           if (robjmsg != null) 
           {
             rmsg = (Message)robjmsg.getObject();
             System.out.print("Name : " +rmsg.getName()) ;
             System.out.print(" Pri: " +robjmsg.getJMSPriority()) ;
             System.out.print(" Message: " +robjmsg.getIntProperty("Id")) ;
             System.out.print(" " +robjmsg.getStringProperty("City")) ;
             System.out.println(" " +robjmsg.getIntProperty("Priority")) ;
           }
           else {
             System.out.println ("No more messages.") ;
             done=true;
           }
           tsess.commit();
         } catch (Exception e) {
           System.out.println("Error in performJmsOperations: " + e) ;
           done=true;
         }
        } /* while loop*/
       }
    } catch (Exception e) {
      System.out.println("Error in performJmsOperations: " + e) ;
      throw e;
    }
 } /* end of demo06 */
}
