/* $Header: aqjmsdemo04.java 16-mar-00.13:05:20 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    03/16/00 - AQ jms demos
    rbhyrava    03/15/00 - AQ JMS demo - Object Type Payload
    rbhyrava    03/15/00 - Creation
 */

/**
 *  @version $Header: aqjmsdemo04.java 16-mar-00.13:05:20 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 Publish/Receive messages to/from topic. The Oracle Type Payload is 
 *  used. Dequeue is based on the Payload content 
 * This is an Oracle Extention to JMS.
 * This demo does the following:
 * -- Create a multi-consumer topic with Payload type as EMP
 *    Refer to aqjmsdemo.sql for definition of EMP and CARS Object Type 
      definitions
 * -- Create a Topic Publisher 
 * -- Create two durable subscribers based on the Payload content 
 * -- Publish messages 
 * -- Receive the messages for both the subscribers 
 *
 * 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 aqjmsdemo04.java  Cars.java Emp.java 
 *
 *  5) java aqjmsdemo04  <SID> <HOST> <PORT> <DRIVER> 
 *     Example usage: 
 *       java aqjmsdemo04 orcl82 dlsun666 1521 thin 
 *
 ***/
/* import useful packages */
import oracle.AQ.*;
import oracle.jms.*;
import javax.jms.*;
import java.lang.*;

public class aqjmsdemo04
{
  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) ;
          publishReceiveMessages(tsess);
          tsess.close() ;
          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 qtprop ;
      AQQueueTable qtable;
      AQjmsDestinationProperty dprop;
      Topic topic;
      try {
         /* Create Queue Tables */
         System.out.println("Creating Queue Table...") ;
         /* Drop the queue if already exists */
         try {
           qtable=((AQjmsSession)tsess).getQueueTable("jmsuser", "jmsadtqueue");
           qtable.drop(true);
         } catch (Exception e) {} ;

         qtprop = new AQQueueTableProperty ("EMP") ;
         qtprop.setComment("EMP queue") ;
         qtprop.setMultiConsumer(true) ;
         qtprop.setCompatible("8.1") ;
         qtprop.setPayloadType("JMSUSER.EMP") ;

         qtable = ((AQjmsSession)tsess).createQueueTable("JMSUSER",
                         "JMSADTQUEUE", qtprop) ;

         System.out.println ("Creating Topic EMPCARS...");
         dprop = new AQjmsDestinationProperty() ;
         dprop.setComment("create topic") ;
         topic= ((AQjmsSession)tsess).createTopic( qtable,"EMPCARS",dprop);

         /* Start the topic */
         ((AQjmsDestination)topic).start(tsess, true, true);
         System.out.println("Successfully setup Topic");  
      } catch (Exception ex) {
         System.out.println("Error in setupTopic: " + ex);
         throw ex;
      }
   }

  public static void publishReceiveMessages(TopicSession tsess) 
     throws Exception
  {
    Topic topic =null;
    TopicSubscriber[] subs;
    TopicPublisher publisher;
    StringBuffer txtbuf ;
    AdtMessage adt_msg;
    Emp empmsg, rmsg;
    Cars c;
    try
    {
    
      /* Get the topic */
      System.out.println("Getting the Topic...") ;
      topic = ((AQjmsSession)tsess).getTopic("JMSUSER", "EMPCARS");

      System.out.println("Creating Topic Subscribers...") ;
      subs = new TopicSubscriber[2] ;

      subs[0] = ((AQjmsSession)tsess).createDurableSubscriber(topic, 
               "PROG1", "TAB.USER_DATA.CAROWN.COLOR = 'RED'", 
                   false, Emp.getFactory()) ;
      subs[1] = ((AQjmsSession)tsess).createDurableSubscriber(topic, 
               "PROG2", "TAB.USER_DATA.RANK > 2", 
                   false, Emp.getFactory()) ;

      System.out.println("Create a Publisher...") ;
      publisher = tsess.createPublisher(topic); 

      adt_msg = ((AQjmsSession)tsess).createAdtMessage() ;

      String[] names={"KING JOE","SCOTT RAY","ADITYA NELLORE","LORI MAXWELL" };
      int[] rank={1,2,3,4} ;
      String color[] = {"YELLOW", "RED", "BLUE","RED" } ;
      String make[] = {"BMW", "ACURA", "MERCEDES","LEXUS" } ;
      empmsg = new Emp() ;
      c = new Cars() ;
  
      System.out.println("Publish messages...") ;

      for ( int i = 1 ; i < 4 ; i++) 
      {
         java.math.BigDecimal id = new java.math.BigDecimal(i) ;
         empmsg.setId(id) ;
         empmsg.setName(names[i]) ;
         empmsg.setRank(new java.math.BigDecimal(rank[i])) ;
         c.setCarno(new java.math.BigDecimal(1000 + i)) ;
         c.setMake(make[i] );
         c.setColor(color[i] );
         c.setPrice(new java.math.BigDecimal(10000 * i) );
         empmsg.setCarown(c) ;

         adt_msg.clearProperties() ;
         //adt_msg.setJMSCorrelationID(new Integer(100+i) ) ;
         adt_msg.setAdtPayload(empmsg) ;
         adt_msg.setJMSPriority(1+(i%3)) ;
         publisher.publish(topic,adt_msg, DeliveryMode.PERSISTENT, 
              1+(i%3),AQjmsConstants.EXPIRATION_NEVER);
         
         System.out.print("Sent Message Pri: " +adt_msg.getJMSPriority()) ;
         System.out.print(" Id: " +empmsg.getId()) ;
         System.out.print(" Name: " +empmsg.getName()) ;
         System.out.print(" Rank: " +empmsg.getRank()) ;
         System.out.print(" Car: " +empmsg.getCarown().getColor()) ;
         System.out.println(" " +empmsg.getCarown().getMake()) ;
      }

      System.out.println("Commit") ;
      tsess.commit() ;

      Thread.sleep(500);

      System.out.println("Receive Messages...") ;

      /* Receive messages for each subscriber */

      for (int i=0; i< subs.length ; i++) 
      {
         System.out.println ("Messages for subscriber : " +i) ;
         boolean more = true;
         while(more) {
           try {
             adt_msg = (AdtMessage)( subs[i].receiveNoWait() );
             if (adt_msg != null) 
             {
               rmsg = (Emp) (adt_msg.getAdtPayload());
               System.out.print("Priority : " +adt_msg.getJMSPriority()) ;
               System.out.print(" Id: " +rmsg.getId()) ;
               System.out.print(" Name: " +rmsg.getName()) ;
               System.out.print(" Rank: " +rmsg.getRank()) ;
               System.out.print(" Car: " +rmsg.getCarown().getColor()) ;
               System.out.println(" " +rmsg.getCarown().getMake()) ;
             }
             else {
               System.out.println ("No more messages.") ;
               more=false;
             }
             tsess.commit();
           } catch (Exception e) {
               System.out.println ("Error in Receive: " + e) ;
               more=false;
           }
        } /* while loop*/
       }
    } catch (Exception e) {
      System.out.println("Error in publish receive Messages: " + e) ;
      throw e;
    }
 }
}
