com/scalagent/kjoram/ConnectionConsumer.java

/*
 * JORAM: Java(TM) Open Reliable Asynchronous Messaging
 * Copyright (C) 2001 - ScalAgent Distributed Technologies
 * Copyright (C) 1996 - Dyade
 *
 * This library is free software; you can redistribute it and/or
 * modify it under the terms of the GNU Lesser General Public
 * License as published by the Free Software Foundation; either
 * version 2.1 of the License, or any later version.
 * 
 * This library is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
 * Lesser General Public License for more details.
 * 
 * You should have received a copy of the GNU Lesser General Public
 * License along with this library; if not, write to the Free Software
 * Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307
 * USA.
 *
 * Initial developer(s): Frederic Maistre (INRIA)
 * Contributor(s): Nicolas Tachker (ScalAgent)
 */
package com.scalagent.kjoram;

import com.scalagent.kjoram.jms.*;
import com.scalagent.kjoram.excepts.*;

import java.util.Vector;
import java.util.Enumeration;


public class ConnectionConsumer
{
  private Connection cnx;
  private boolean durable = false;
  private String selector;
  private ServerSessionPool sessionPool;
  private int maxMessages;
  private CCDaemon ccDaemon;
  private com.scalagent.kjoram.jms.AbstractJmsRequest currentReq = null;
  private boolean closed = false;

  String targetName;
  boolean queueMode = true;
  com.scalagent.kjoram.util.Queue repliesIn;


  ConnectionConsumer(Connection cnx, Destination dest, String subName,
                     String selector, ServerSessionPool sessionPool,
                     int maxMessages) throws JMSException
  {
    if (sessionPool == null)
      throw new JMSException("Invalid ServerSessionPool parameter: "
                             + sessionPool);
    if (maxMessages <= 0)
      throw new JMSException("Invalid maxMessages parameter: " + maxMessages);
    
    this.cnx = cnx;
    this.selector = selector;
    this.sessionPool = sessionPool;
    this.maxMessages = maxMessages;

    if (dest instanceof Queue)
      targetName = dest.getName();
    else if (subName == null) {
      queueMode = false;
      targetName = cnx.nextSubName();
    }
    else {
      queueMode = false;
      targetName = subName;
      durable = true;
    }

    repliesIn = new com.scalagent.kjoram.util.Queue();

    if (cnx.cconsumers == null)
      cnx.cconsumers = new Vector();
 
    cnx.cconsumers.addElement(this);

    ccDaemon = new CCDaemon(this);
    ccDaemon.setDaemon(true);
    ccDaemon.start();

    // If the consumer is a subscriber, subscribing to the target topic:
    if (! queueMode) 
      cnx.syncRequest(new ConsumerSubRequest(dest.getName(), targetName,
                                             selector, false, durable));

    // Sending a listener request:
    currentReq = new ConsumerSetListRequest(targetName, selector, queueMode);
    currentReq.setRequestId(cnx.nextRequestId());
    cnx.requestsTable.put(currentReq.getKey(), this);
    cnx.asyncRequest(currentReq);

    if (JoramTracing.dbgClient)
      JoramTracing.log(JoramTracing.DEBUG, this + ": created.");
  }

  ConnectionConsumer(Connection cnx, Destination dest, String selector,
                     ServerSessionPool sessionPool,
                     int maxMessages) throws JMSException
  {
    this(cnx, dest, null, selector, sessionPool, maxMessages);
  }

  public String toString()
  {
    return "ConnCons:" + cnx.toString();
  }


  public ServerSessionPool getServerSessionPool() throws JMSException
  {
    if (closed)
      throw new com.scalagent.kjoram.excepts.
        IllegalStateException("Forbidden call on a closed"
                              + " ConnectionConsumer.");
    return sessionPool;
  }


  public void close() throws JMSException
  {
    cnx.requestsTable.remove(currentReq.getKey());
    ccDaemon.stop();

    // If the consumer is a subscriber, managing the subscription closing: 
    if (! queueMode) {
      try {
        if (durable) 
          cnx.syncRequest(new ConsumerCloseSubRequest(targetName));
        else
          cnx.syncRequest(new ConsumerUnsubRequest(targetName));
      }
      // A JMSException might be caught if the connection is broken.
      catch (JMSException jE) {}
    }
    cnx.cconsumers.removeElement(this);
  }

class CCDaemon extends com.scalagent.kjoram.util.Daemon
{
  private ConnectionConsumer cc;

  CCDaemon(ConnectionConsumer cc)
  {
    super(cc.toString());
    this.cc = cc;
  }

  public void run()
  {
    ConsumerMessages reply;
    Vector deliveries = new Vector();
    ServerSession serverSess;
    Session sess;
    int counter;

    try {
      while (running) {
        canStop = true; 

        try {
          // Expecting a reply:
          repliesIn.get();
        }
        catch (Exception iE) {
          continue;
        }
        canStop = false;

        // Processing the delivery:
        try {
          if (JoramTracing.dbgClient)
            JoramTracing.log(JoramTracing.DEBUG, "--- " + cc
                             + ": got a delivery.");

          // Getting a server's session:
          serverSess = sessionPool.getServerSession();
          sess = (Session) serverSess.getSession();
          sess.connectionConsumer = cc;
          counter = 1;

          // As long as there are messages to deliver, passing to session(s)
          // as many messages as possible:
          while (counter <= maxMessages && repliesIn.size() > 0) {
            
            // If the consumer is a queue consumer, sending a new request:
            if (queueMode) {
              cnx.requestsTable.remove(currentReq.getKey());
              currentReq = new ConsumerSetListRequest(targetName, selector,
                                                      queueMode);
              currentReq.setRequestId(cnx.nextRequestId());
              cnx.requestsTable.put(currentReq.getKey(), cc);
              cnx.asyncRequest(currentReq);
            }

            reply = (ConsumerMessages) repliesIn.pop();
            for (Enumeration e = reply.getMessages().elements(); e.hasMoreElements(); ) {
              deliveries.addElement(e.nextElement());
            }

            while (! deliveries.isEmpty()) {
              while (counter <= maxMessages && ! deliveries.isEmpty()) {
                if (JoramTracing.dbgClient)
                  JoramTracing.log(JoramTracing.DEBUG, "Passes a"
                                   + " message to a session.");
                Object obj = deliveries.elementAt(0);
                deliveries.removeElementAt(0);
                sess.repliesIn.push(obj);
                counter++;
              }
              if (counter > maxMessages) {
                if (JoramTracing.dbgClient)
                  JoramTracing.log(JoramTracing.DEBUG, "Starts the"
                                   + " session.");
                serverSess.start(); 
                counter = 1;

                if (! deliveries.isEmpty() || repliesIn.size() > 0) {
                  serverSess = sessionPool.getServerSession();
                  sess =
                    (Session) serverSess.getSession();
                  sess.connectionConsumer = cc;
                }
              }
            }
          }
          // There is no more message to deliver and no more delivery, 
          // starting the last session to which messages have been passed:
          if (JoramTracing.dbgClient)
            JoramTracing.log(JoramTracing.DEBUG, "No more delivery.");
          if (counter > 1) {
            if (JoramTracing.dbgClient)
              JoramTracing.log(JoramTracing.DEBUG, "Starts the"
                               + " session.");
            counter = 1;
            serverSess.start();
          }
        }
        // A JMSException will be caught if the application server failed
        // to provide a session: closing the consumer.
        catch (JMSException jE) {
          canStop = true;
          try {
            cc.close();
          }
          catch (JMSException jE2) {}
        }
      }
    }
    finally {
      finish();
    }
  }

  public void shutdown()
  {}

  public void close()
  {
    if (JoramTracing.dbgClient)
      JoramTracing.log(JoramTracing.DEBUG, "CCDaemon finished.");
  }
}
}

---