import java.io.File;
import java.io.FileOutputStream;
import java.io.IOException;
import java.io.PrintStream;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.sql.Statement;
import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.Hashtable;
import java.util.Random;

import com.ibm.mq.MQException;
import com.ibm.mq.MQGetMessageOptions;
import com.ibm.mq.MQMessage;
import com.ibm.mq.MQQueue;
import com.ibm.mq.MQQueueManager;
import com.ibm.mq.constants.CMQC;

/**
 * Program Name
 *  BEServer01
 *
 * Description
 *  This java class will:
 *  - connect to a remote queue manager
 *  - open request queue
 *  - start thread to 
 *  - - open temporary dynamic queue for COA & COD messages
 *  - - loop until main task stops 
 *  - - - Get COA & COD messages
 *  - - - update DB
 *  - - close queue
 *  - - disconnect from queue manager
 *  - loop until 2 minutes without a request message  
 *  - - get a request message
 *  - - insert record to DB
 *  - - put a reply message 
 *  - - update record in DB
 *  - close queue
 *  - disconnect from queue manager
 *  - close DB 
 *
 * Sample Command Line Parameters
 *  -m MQA1 -h 127.0.0.1 -p 1414 -c TEST.CHL -q REQUEST.Q -u userId -x password
 *
 * @author Roger Lacroix, Capitalware Inc.
 */
public class BEServer01
{
   private static final String            HEX = "0123456789ABCDEF";
   private static final SimpleDateFormat  lOGGER_TIMESTAMP = new SimpleDateFormat("yyyy/MM/dd HH:mm:ss.SSS");

   private Hashtable<String, String>   params = null;
   private Hashtable<String, Object>   mqht;
   private String                      qMgrName;
   private String                      requestQName;
   private static PrintStream          ps = null;
   private static boolean              debugMe = false;

   /**
    * The constructor
    * @param args
    * @throws IllegalArgumentException
    * @throws ClassNotFoundException
    */
   public BEServer01(String[] args) throws IllegalArgumentException, ClassNotFoundException
   {
      super();
//      writeLogToFile();  // have logging info go to a file rather the console.

      if (debugMe) BEServer01.logger("");

      BEServer01.logger("Starting");

      initParms(args);

      // load the sqlite-JDBC driver using the current class loader
      Class.forName("org.sqlite.JDBC");
      String dbName = "BEServer01_"+qMgrName+".db";
      String dbURL = "jdbc:sqlite:" + dbName;
      if (!((new File(dbName)).exists()))
         createNewTable(dbURL);
      begin(dbURL);
   }

   /**
    * Extract the command-line parameters and initialize the MQ HashTable.
    * @param args
    * @throws IllegalArgumentException
    */
   private void initParms(String[] args) throws IllegalArgumentException
   {
      if (debugMe) BEServer01.logger("");
      
      int port = 1414;
      params = new Hashtable<String, String>(10);
      if (args.length > 0 && (args.length % 2) == 0)
      {
         for (int i = 0; i < args.length; i += 2)
         {
            params.put(args[i], args[i + 1]);
         }
      }
      else
      {
         throw new IllegalArgumentException();
      }

      if (allParamsPresent())
      {
         qMgrName = (String) params.get("-m");
         requestQName = (String) params.get("-q");
         
         try
         {
            port = Integer.parseInt((String) params.get("-p"));
         }
         catch (NumberFormatException e)
         {
            port = 1414;
         }

         mqht = new Hashtable<String, Object>();
         
         mqht.put(CMQC.CHANNEL_PROPERTY, params.get("-c"));
         mqht.put(CMQC.HOST_NAME_PROPERTY, params.get("-h"));
         mqht.put(CMQC.PORT_PROPERTY, new Integer(port));
         mqht.put(CMQC.USER_ID_PROPERTY, params.get("-u"));
         mqht.put(CMQC.PASSWORD_PROPERTY, params.get("-x"));

         // I don't want to see MQ exceptions at the console.
         MQException.log = null;
      }
      else
      {
         throw new IllegalArgumentException();
      }
   }

   /**
    * Make sure the required parameters are present.
    * @return true/false
    */
   private boolean allParamsPresent()
   {
      if (debugMe) BEServer01.logger("");
      
      boolean b = params.containsKey("-h") && params.containsKey("-p") &&
                  params.containsKey("-c") && params.containsKey("-m") &&
                  params.containsKey("-q") && 
                  params.containsKey("-u") && params.containsKey("-x");
      if (b)
      {
         try
         {
            Integer.parseInt((String) params.get("-p"));
         }
         catch (NumberFormatException e)
         {
            b = false;
         }
      }

      return b;
   }

   /**
    * Connect, open queues, loop, close queues and disconnect.
    * @param dbUrl
    */
   private void begin(String dbURL)
   {
      if (debugMe) BEServer01.logger("");
      
      Connection dbConn = null;
      MQQueueManager qMgr = null;
      MQQueue requestQ = null;
      int inputOpenOptions = CMQC.MQOO_INQUIRE + CMQC.MQOO_INPUT_SHARED + CMQC.MQOO_FAIL_IF_QUIESCING;

      try
      {
         // create a database connection
         dbConn = DriverManager.getConnection(dbURL);
      
         COGetHandler cogh = new COGetHandler(dbConn);
         String coQName = cogh.getCOQName().trim(); /* Temp. Dynamic Queue name to receive COA & COD messages */
         
         BEServer01.logger("CO reply Q name "+ coQName);
         
         if ( (coQName != null) && !("".equals(coQName)) )
         {
            cogh.start();  // Start the thread to handle receiving the COA & COD messages
            
            try
            {
               qMgr = new MQQueueManager(qMgrName, mqht);
               BEServer01.logger("successfully connected to "+ qMgrName);
   
               requestQ = qMgr.accessQueue(requestQName, inputOpenOptions);
               BEServer01.logger("successfully opened "+ requestQName);
               
               getRequestMessages(qMgr, requestQ, coQName, dbConn);
            }
            catch (MQException e)
            {
               BEServer01.logger("cc=" +e.completionCode + " : rc=" + e.reasonCode);
            }
            finally
            {
               cogh.setStop();
   
               if (requestQ != null)
               {
                  try
                  {
                     requestQ.close();
                     BEServer01.logger("closed: "+ requestQName);
                  }
                  catch (MQException e)
                  {
                     BEServer01.logger("close cc=" +e.completionCode + " : rc=" + e.reasonCode);
                  }
               }
               
               if ( qMgr != null)
               {
                  try
                  {
                     qMgr.disconnect();
                     BEServer01.logger("disconnected from "+ qMgrName);
                  }
                  catch (MQException e)
                  {
                     BEServer01.logger("disconnect cc=" +e.completionCode + " : rc=" + e.reasonCode);
                  }
               }
            }
         }
      }
      catch (SQLException es)
      {
         BEServer01.logger(es.getLocalizedMessage());
      }
   }

   /**
    * Get up to 2 messages: COA & COD from the input queue
    * Wait up to 10 seconds for each message.
    * @param qMgr
    * @param inQ
    * @param coQName
    * @param dbConn
    */
   private void getRequestMessages(MQQueueManager qMgr, MQQueue inQ, String coQName, Connection dbConn)
   {
      if (debugMe) BEServer01.logger("");
      
      MQGetMessageOptions gmo = new MQGetMessageOptions();
      gmo.options = CMQC.MQGMO_FAIL_IF_QUIESCING + CMQC.MQGMO_WAIT;
      gmo.waitInterval = 60 * 1000; // wait up to 60 seconds

      MQMessage requestMsg = null;
      byte[] b;
      int mCount = 1;
      boolean more = true;

      while (more)
      {
         requestMsg = new MQMessage();
         requestMsg.messageId = CMQC.MQMI_NONE;
         requestMsg.correlationId = CMQC.MQCI_NONE;

         try
         {
            inQ.get(requestMsg, gmo);

            /**
             * Did the user tell us to exit?
             */
            if (requestMsg.feedback == CMQC.MQFB_QUIT)
            {
               more = false;
               BEServer01.logger("Quit message received.");
               break;
            }
            
            Date received_TS = new Date();
            String msgIdAsStr = bytesToHex(requestMsg.messageId);
            
            b = new byte[requestMsg.getMessageLength()];
            requestMsg.readFully(b);
            BEServer01.logger("Received request message: \"" + new String(b) + "\"");

            // Insert record in table
            insertToTable(dbConn, msgIdAsStr, received_TS, requestMsg.putDateTime.getTime());
            
            /**
             * Go do whatever business logic needs to be done.
             */
            
            /**
             * Fake a business process 
             * Sleep for a random # of milliseconds between 100 & 300
             */
            try 
            {
               Random rand = new Random();
               int sleepMS = 100 + rand.nextInt(200);
               Thread.sleep(sleepMS);
            }
            catch(Exception ee) 
            {}
            
            /**
             * Skip the 4th message to fake a message getting lost.
             */
            if ( mCount != 4)
            {
               sendReplyMessage(qMgr, 
                                requestMsg.replyToQueueManagerName.trim(), 
                                requestMsg.replyToQueueName.trim(), 
                                coQName, 
                                requestMsg.messageId, 
                                msgIdAsStr,
                                mCount, 
                                dbConn);
            }
            
            mCount++;
         }
         catch (MQException e)
         {
            BEServer01.logger("cc=" +e.completionCode + " : rc=" + e.reasonCode);
            more = false;
         }
         catch (IOException e)
         {
            BEServer01.logger(e.getLocalizedMessage());
            more = false;
         }
      }
   }

   /**
    * Send a reply message to the requesting client.
    * @param qMgr
    * @param replyQMgr
    * @param replyQ
    * @param coQName
    * @param messageId
    * @param msgIdAsStr
    * @param mCount
    * @param dbConn
    */
   private void sendReplyMessage(MQQueueManager qMgr, String replyQMgr, String replyQ, String coQName, byte[] messageId, String msgIdAsStr, int mCount,  Connection dbConn)
   {
      if (debugMe) BEServer01.logger("");

      try
      {
         String replyText = "This is reply message # "+mCount++;
         MQMessage replyMsg = new MQMessage();
         replyMsg.messageId = CMQC.MQMI_NONE;
         replyMsg.correlationId = messageId;
         replyMsg.messageType = CMQC.MQMT_REPLY;
         replyMsg.replyToQueueManagerName = qMgrName;
         replyMsg.replyToQueueName = coQName;
         replyMsg.report = CMQC.MQRO_COA | CMQC.MQRO_COD;
         replyMsg.writeString(replyText);

         // Now send a reply message
         qMgr.put(replyQ, replyQMgr, replyMsg);
         BEServer01.logger("Sent reply message: \"" + replyText + "\"" + " to " + replyQ + " of " + replyQMgr);

         // Update record in table
         updateTable(dbConn, msgIdAsStr, bytesToHex(replyMsg.messageId), replyMsg.putDateTime.getTime());
      }
      catch (MQException e)
      {
         BEServer01.logger("cc=" +e.completionCode + " : rc=" + e.reasonCode);
      }
      catch (IOException e)
      {
         BEServer01.logger(e.getLocalizedMessage());
      }
   }

   /**
    * Create new table in DB.
    * @param dbURL
    */
   private void createNewTable(String dbURL)
   {
      if (debugMe) BEServer01.logger("");
      
      Connection dbConn = null;
      
      try
      {
         // create a database connection
         dbConn = DriverManager.getConnection(dbURL);
         Statement statement = dbConn.createStatement();

         statement.executeUpdate("CREATE TABLE 'MQSLA' ( " +
                                                       "   'MessageId'      TEXT NOT NULL PRIMARY KEY UNIQUE," +
                                                       "   'Received_TS'    TEXT," +
                                                       "   'Request_TS'     TEXT," +
                                                       "   'Reply_TS'       TEXT," +
                                                       "   'ReplyMessageId' TEXT KEY UNIQUE," +
                                                       "   'COA_TS'         TEXT," +
                                                       "   'COD_TS'         TEXT"  +
                                                       ");");

         BEServer01.logger("MQSLA table created.");
      }
      catch (SQLException e)
      {
         // if the error message is "out of memory", it probably means no database file is found
         BEServer01.logger(e.getLocalizedMessage());
      }
      finally
      {
         try
         {
            if (dbConn != null)
               dbConn.close();
         }
         catch (SQLException e)
         {
            BEServer01.logger(e.getLocalizedMessage());
         }
      }
   }


   /**
    * Insert a new row into the table
    * @param dbConn
    * @param msgId
    * @param received_TS
    * @param request_TS
    */
   private void insertToTable(Connection dbConn, String msgId, Date received_TS, Date request_TS)
   {
      if (debugMe) BEServer01.logger("");
      
      SimpleDateFormat dtFormat = new SimpleDateFormat("HH:mm:ss.SSS");
      PreparedStatement pStmt = null;
      PreparedStatement pStmt2 = null;
      
      try
      {
         pStmt = dbConn.prepareStatement("INSERT INTO MQSLA (MessageId,received_TS,request_TS) VALUES (?, ?, ?)");
         pStmt.setString(1, msgId);
         pStmt.setString(2, dtFormat.format(received_TS));
         pStmt.setString(3, dtFormat.format(request_TS));

         int rcode = pStmt.executeUpdate();
         RQClient01.logger("Insert rcode="+rcode);

         pStmt2 = dbConn.prepareStatement("SELECT MessageId,received_TS,request_TS FROM MQSLA WHERE MessageId = ?");
         pStmt2.setString(1, msgId);

         ResultSet rs = pStmt2.executeQuery();
         while (rs.next())
         {
            // read the result set
            BEServer01.logger("MessageId    = " + rs.getString("MessageId"));
            BEServer01.logger("received_TS  = " + rs.getString("received_TS"));
            BEServer01.logger("request_TS   = " + rs.getString("request_TS"));
         }
      }
      catch (SQLException e)
      {
         BEServer01.logger(e.getLocalizedMessage());
      }
      finally
      {
         closePStmt(pStmt);
         closePStmt(pStmt2);
      }
   }

   /**
    * Update an existing row in the table 
    * @param dbConn
    * @param msgId
    * @param colName
    * @param reply_TS
    */
   private void updateTable(Connection dbConn, String msgId, String replyMsgId, Date reply_TS)
   {
      if (debugMe) BEServer01.logger("");
      
      SimpleDateFormat dtFormat = new SimpleDateFormat("HH:mm:ss.SSS");
      PreparedStatement pStmt = null;
      PreparedStatement pStmt2 = null;
      
      try
      {
         pStmt = dbConn.prepareStatement("UPDATE MQSLA SET Reply_TS= ?, ReplyMessageId = ? WHERE MessageId = ?");
         pStmt.setString(1, dtFormat.format(reply_TS));
         pStmt.setString(2, replyMsgId);
         pStmt.setString(3, msgId);
         
         int rcode = pStmt.executeUpdate();
         BEServer01.logger("Update rcode="+rcode);

         pStmt2 = dbConn.prepareStatement("SELECT MessageId,received_TS,request_TS,Reply_TS,ReplyMessageId FROM MQSLA WHERE MessageId = ?");
         pStmt2.setString(1, msgId);

         ResultSet rs = pStmt2.executeQuery();
         while (rs.next())
         {
            // read the result set
            BEServer01.logger("MessageId      = " + rs.getString("MessageId"));
            BEServer01.logger("Received_TS    = " + rs.getString("Received_TS"));
            BEServer01.logger("Request_TS     = " + rs.getString("Request_TS"));
            BEServer01.logger("Reply_TS       = " + rs.getString("Reply_TS"));
            BEServer01.logger("ReplyMessageId = " + rs.getString("ReplyMessageId"));
         }
      }
      catch (SQLException e)
      {
         BEServer01.logger(e.getLocalizedMessage());
      }
      finally
      {
         closePStmt(pStmt);
         closePStmt(pStmt2);
      }
   }

   /**
    * CLose the Prepared Statement
    * @param pStmt
    */
   private void closePStmt(PreparedStatement pStmt)
   {
      try
      {
         if (pStmt != null)
            pStmt.close();
      }
      catch (SQLException e)
      {
         BEServer01.logger(e.getLocalizedMessage());
      }
   }

   /**
    * Convenience method to convert a byte array to a hex string.
    *
    * @param data the byte[] to convert
    * @return String the converted byte[]
    */
   private String bytesToHex(byte[] data)
   {
      if (debugMe) BEServer01.logger("");
      
      StringBuffer buf = new StringBuffer();
      for (int i = 0; i < data.length; i++)
      {
         buf.append(byteToHex(data[i]));
      }
      return buf.toString();
   }

   /**
    * Convenience method to convert a byte to a hex string.
    *
    * @param data the byte to convert
    * @return String the converted byte
    */
   private String byteToHex(byte data)
   {
      int hi = (data & 0xF0) >> 4;
      int lo = (data & 0x0F);
      return "" + HEX.charAt(hi) + HEX.charAt(lo);
   }

   /**
    * Code to setup to have logging information go to a file rather than console.
    */
   private void writeLogToFile()
   {
      if (debugMe) BEServer01.logger("");
      
      try
      {
         String logFile = "BEServer01.log";

         if ((new File(logFile)).length() > 1048576)                     // Larger than 1MB
            ps = new PrintStream(new FileOutputStream(logFile, false));  // create new
         else
            ps = new PrintStream(new FileOutputStream(logFile, true));   // append
      }
      catch(Exception e)
      {
         BEServer01.logger(e.getLocalizedMessage());
      }

      if (ps != null)
      {
         System.setOut(ps);
         System.setErr(ps);
      }
   }

   /**
    * A simple logger method
    * @param data
    */
   public static void logger(String data)
   {
      String className = Thread.currentThread().getStackTrace()[2].getClassName();

      // Remove the package info.
      if ( (className != null) && (className.lastIndexOf('.') != -1) )
         className = className.substring(className.lastIndexOf('.')+1);

      System.out.println(lOGGER_TIMESTAMP.format(new Date())+" "+className+": "+Thread.currentThread().getStackTrace()[2].getMethodName()+": "+data);
   }

   /**
    * main line
    * @param args
    */
   public static void main(String[] args)
   {
      if (System.getProperty("BEServer01_DEBUG") != null)
         debugMe = true;

      try
      {
         new BEServer01(args);
      }
      catch (ClassNotFoundException e)
      {
         BEServer01.logger(e.getLocalizedMessage());
      }
      catch (IllegalArgumentException e)
      {
         System.err.println("Usage: java BEServer01 -m QueueManagerName -h host -p port -c channel -q RequestQueueName -u userId -x password");
      }
      
      BEServer01.logger("Ending");
   }

   /**
    * Class to handle getting the COA & COD messages in a separate thread. 
    */
   class COGetHandler extends Thread
   {
      private MQQueueManager  qMgr = null;
      private MQQueue         inQ = null;
      private String          coQName = ""; /* Queue name to receive COA & COD messages */
      private boolean         running = true;
      private Connection      dbConn = null;

      /**
       * The constructor
       * @param dbConn
       */
      public COGetHandler(Connection dbConn)
      {
         super();
         if (debugMe) BEServer01.logger("");
         
         this.dbConn = dbConn;

         int inputOpenOptions = CMQC.MQOO_INQUIRE + CMQC.MQOO_INPUT_SHARED + CMQC.MQOO_FAIL_IF_QUIESCING;
         String modelQName    = "SYSTEM.DEFAULT.MODEL.QUEUE";
         String dynamicQName  = "AMQ.BEServer01.*";

         try
         {
            qMgr = new MQQueueManager(qMgrName, mqht);
            BEServer01.logger("successfully connected to "+ qMgrName);
            
            /* Create Temporary Dynamic Queue to receive COA & COD messages*/
            inQ = qMgr.accessQueue(modelQName, inputOpenOptions, null, dynamicQName, null); 

            coQName = inQ.getName();
            BEServer01.logger("successfully opened "+ coQName);
         }
         catch (MQException e)
         {
            BEServer01.logger("cc=" +e.completionCode + " : rc=" + e.reasonCode);
            
            if (inQ != null)
            {
               try
               {
                  inQ.close();
                  BEServer01.logger("closed: "+ coQName);
               }
               catch (MQException e2)
               {
                  BEServer01.logger("close cc=" +e2.completionCode + " : rc=" + e2.reasonCode);
               }
            }
            
            if ( qMgr != null)
            {
               try
               {
                  qMgr.disconnect();
                  BEServer01.logger("disconnected from "+ qMgrName);
               }
               catch (MQException e2)
               {
                  BEServer01.logger("disconnect cc=" +e2.completionCode + " : rc=" + e2.reasonCode);
               }
            }
         }
      }
      
      /**
       * Worker thread
       */
      public void run()
      {
         if (debugMe) BEServer01.logger("");
         
         MQMessage coMsg = null;
         MQGetMessageOptions gmo = new MQGetMessageOptions();
         gmo.options = CMQC.MQGMO_FAIL_IF_QUIESCING + CMQC.MQGMO_WAIT;
         /* Wait for 2 seconds that way we do not hold up stopping the thread */
         gmo.waitInterval = 2 * 1000; // 2 seconds

         while (running)
         {
            coMsg = new MQMessage();
            coMsg.messageId = CMQC.MQMI_NONE;
            coMsg.correlationId = CMQC.MQCI_NONE;
            
            try
            {
               if (debugMe) BEServer01.logger("Attempting to get a message.");
               inQ.get(coMsg, gmo);
               
               /**
                * Note: The original MessageId will now be in the CorrelationId
                */
               String replyMsgIdAsStr = bytesToHex(coMsg.correlationId);
               
               if ( (coMsg.messageType == CMQC.MQMT_REPORT) && 
                    (coMsg.feedback == CMQC.MQFB_COA) )
               {
                  BEServer01.logger("Received COA");
                   
                  /**
                   * This thread could get ahead of the mothership in updating
                   * the DB record.  So put in a small delay.  
                   */
                  try 
                  {
                     Thread.sleep(75);  
                  }
                  catch(Exception ee) 
                  {}

                  // Update record in table
                  updateTableForReply(dbConn, replyMsgIdAsStr, "COA_TS", coMsg.putDateTime.getTime());
               }
               else if ( (coMsg.messageType == CMQC.MQMT_REPORT) && 
                         (coMsg.feedback == CMQC.MQFB_COD) )
               {
                  BEServer01.logger("Received COD");

                  // Update record in table
                  updateTableForReply(dbConn, replyMsgIdAsStr, "COD_TS", coMsg.putDateTime.getTime());
               }
               else
               {
                  BEServer01.logger("Unknown message type: " + coMsg.messageType);
               }
            }
            catch (MQException e)
            {
               if ( (e.completionCode == CMQC.MQCC_FAILED) && (e.reasonCode == CMQC.MQRC_NO_MSG_AVAILABLE) )
               {
                  /* no message on the queue - loop again */
               }
               else
               {
                  running = false;
                  BEServer01.logger("close cc=" +e.completionCode + " : rc=" + e.reasonCode);
               }
            }
         }

         if (inQ != null)
         {
            try
            {
               inQ.close();
               BEServer01.logger("closed: "+ coQName);
            }
            catch (MQException e2)
            {
               BEServer01.logger("close cc=" +e2.completionCode + " : rc=" + e2.reasonCode);
            }
         }
         
         if ( qMgr != null)
         {
            try
            {
               qMgr.disconnect();
               BEServer01.logger("disconnected from "+ qMgrName);
            }
            catch (MQException e2)
            {
               BEServer01.logger("disconnect cc=" +e2.completionCode + " : rc=" + e2.reasonCode);
            }
         }
      }

      /**
       * Update an existing row in the table 
       * @param dbConn
       * @param replyMsgId
       * @param colName
       * @param colValue
       */
      private void updateTableForReply(Connection dbConn, String replyMsgId, String colName, Date colValue)
      {
         if (debugMe) BEServer01.logger("");
         
         SimpleDateFormat dtFormat = new SimpleDateFormat("HH:mm:ss.SSS");
         PreparedStatement pStmt = null;
         PreparedStatement pStmt2 = null;

         try
         {
            pStmt = dbConn.prepareStatement("UPDATE MQSLA SET "+colName+" = ? WHERE ReplyMessageId = ?");
            pStmt.setString(1, dtFormat.format(colValue));
            pStmt.setString(2, replyMsgId);
            
            int rcode = pStmt.executeUpdate();
            BEServer01.logger("Update rcode="+rcode);

            pStmt2 = dbConn.prepareStatement("SELECT * FROM MQSLA WHERE ReplyMessageId = ?");
            pStmt2.setString(1, replyMsgId);

            ResultSet rs = pStmt2.executeQuery();
            while (rs.next())
            {
               // read the result set
               BEServer01.logger("MessageId      = " + rs.getString("MessageId"));
               BEServer01.logger("Received_TS    = " + rs.getString("Received_TS"));
               BEServer01.logger("Request_TS     = " + rs.getString("Request_TS"));
               BEServer01.logger("Reply_TS       = " + rs.getString("Reply_TS"));
               BEServer01.logger("ReplyMessageId = " + rs.getString("ReplyMessageId"));
               BEServer01.logger("COA_TS         = " + rs.getString("COA_TS"));
               BEServer01.logger("COD_TS         = " + rs.getString("COD_TS"));
            }
         }
         catch (SQLException e)
         {
            BEServer01.logger(e.getLocalizedMessage());
         }
         finally
         {
            closePStmt(pStmt);
            closePStmt(pStmt2);
         }
      }

      /**
       * CLose the Prepared Statement
       * @param pStmt
       */
      private void closePStmt(PreparedStatement pStmt)
      {
         try
         {
            if (pStmt != null)
               pStmt.close();
         }
         catch (SQLException e)
         {
            BEServer01.logger(e.getLocalizedMessage());
         }
      }
      
      /**
       * Get the name of the reply queue
       * @return coQName
       */
      public String getCOQName()
      {
         if (debugMe) BEServer01.logger("");
         
         return coQName;
      }

      /**
       * The mothership will call us when it wants us to stop.
       */
      public void setStop()
      {
         if (debugMe) BEServer01.logger("");
         
         running = false;
      }
   }
}
