com/scalagent/kjoram/Session.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.util.TimerTask;

import java.util.*;

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


public class Session
{
  public static final int SESSION_TRANSACTED = 0;
  public static final int AUTO_ACKNOWLEDGE = 1;
  public static final int CLIENT_ACKNOWLEDGE = 2;
  public static final int DUPS_OK_ACKNOWLEDGE = 3;
  
  private TimerTask closingTask = null;
  private boolean scheduled = false;

  private com.scalagent.kjoram.util.Timer consumersTimer = null;

  protected MessageListener messageListener = null;

  String ident;
  Connection cnx;
  boolean transacted;
  int acknowledgeMode;
  boolean closed = false;
  boolean started = false;

  boolean autoAck;

  Vector consumers;
  Vector producers;
  Vector browsers;
  com.scalagent.kjoram.util.Queue repliesIn;
  SessionDaemon daemon = null;
  int msgListeners = 0;
  Hashtable sendings;
  Hashtable deliveries;

  ConnectionConsumer connectionConsumer = null;


  Session(Connection cnx, boolean transacted,
          int acknowledgeMode) throws JMSException
  {
    if (! transacted 
        && acknowledgeMode != Session.AUTO_ACKNOWLEDGE
        && acknowledgeMode != Session.CLIENT_ACKNOWLEDGE
        && acknowledgeMode != Session.DUPS_OK_ACKNOWLEDGE)
      throw new JMSException("Can't create a non transacted session with an"
                             + " invalid acknowledge mode.");

    this.ident = cnx.nextSessionId();
    this.cnx = cnx;
    this.transacted = transacted;
    this.acknowledgeMode = acknowledgeMode;

    autoAck = ! transacted
              && acknowledgeMode != Session.CLIENT_ACKNOWLEDGE;

    consumers = new Vector();
    producers = new Vector();
    browsers = new Vector();
    repliesIn = new com.scalagent.kjoram.util.Queue();
    sendings = new Hashtable();
    deliveries = new Hashtable();

    // If the session is transacted and the transactions limited by a timer,
    // a closing task might be useful.
    if (transacted && cnx.factoryParameters.txPendingTimer != 0)
      closingTask = new SessionCloseTask();

    cnx.sessions.addElement(this);

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

  public String toString()
  {
    return "Sess:" + ident;
  }


  public int getAcknowledgeMode() throws JMSException
  {
    return acknowledgeMode;
  }

  public boolean getTransacted() throws JMSException
  {
    return transacted;
  }

  public void setMessageListener(MessageListener messageListener)
              throws JMSException
  {
    this.messageListener = messageListener;
  }

  public MessageListener getMessageListener() throws JMSException
  {
    return messageListener;
  }

  public Message createMessage() throws JMSException
  {
    if (closed)
      throw new IllegalStateException("Forbidden call on a closed session.");
    
    return new Message();
  }

  public TextMessage createTextMessage() throws JMSException
  {
    if (closed)
      throw new IllegalStateException("Forbidden call on a closed session.");
    
    return new TextMessage();
  }

  public TextMessage createTextMessage(String text)
         throws JMSException
  {
    if (closed)
      throw new IllegalStateException("Forbidden call on a closed session.");
   
    TextMessage message =  new TextMessage();
    message.setText(text);
    return message;
  }

  public BytesMessage createBytesMessage()
         throws JMSException
  {
    if (closed)
      throw new IllegalStateException("Forbidden call on a closed session.");
    
    return new BytesMessage();
  }

  public MapMessage createMapMessage()
         throws JMSException
  {
    if (closed)
      throw new IllegalStateException("Forbidden call on a closed session.");

    return new MapMessage();
  }

  public QueueBrowser
         createBrowser(Queue queue, String selector)
         throws JMSException
  {
    if (closed)
      throw new IllegalStateException("Forbidden call on a closed session.");

    return new QueueBrowser(this, (Queue) queue, selector);
  }

  public QueueBrowser createBrowser(Queue queue)
         throws JMSException
  {
    if (closed)
      throw new IllegalStateException("Forbidden call on a closed session.");

    return new QueueBrowser(this, (Queue) queue, null);
  }

  public MessageProducer createProducer(Destination dest)
         throws JMSException
  {
    if (closed)
      throw new IllegalStateException("Forbidden call on a closed session.");

    return new MessageProducer(this, (Destination) dest);
  }

  public MessageConsumer
         createConsumer(Destination dest, String selector,
                        boolean noLocal) throws JMSException
  {
    if (closed)
      throw new IllegalStateException("Forbidden call on a closed session.");

    return new MessageConsumer(this, (Destination) dest, selector, null,
                               noLocal);
  }

  public MessageConsumer
         createConsumer(Destination dest, String selector)
         throws JMSException
  {
    if (closed)
      throw new IllegalStateException("Forbidden call on a closed session.");

    return new MessageConsumer(this, (Destination) dest, selector);
  }

  public MessageConsumer createConsumer(Destination dest)
         throws JMSException
  {
    if (closed)
      throw new IllegalStateException("Forbidden call on a closed session.");

    return new MessageConsumer(this, (Destination) dest, null);
  }

  public TopicSubscriber
         createDurableSubscriber(Topic topic, String name,
                                 String selector,
                                 boolean noLocal) throws JMSException
  {
    if (closed)
      throw new IllegalStateException("Forbidden call on a closed session.");

    return new TopicSubscriber(this, (Topic) topic, name, selector, noLocal);
  }

  public TopicSubscriber
         createDurableSubscriber(Topic topic, String name)
         throws JMSException
  {
    if (closed)
      throw new IllegalStateException("Forbidden call on a closed session.");

    return new TopicSubscriber(this, (Topic) topic, name, null, false);
  }

  public Queue createQueue(String queueName) throws JMSException
  {
    if (closed)
      throw new IllegalStateException("Forbidden call on a closed session.");

    return new Queue(queueName);
  }

  public Topic createTopic(String topicName) throws JMSException
  {
    if (closed)
      throw new IllegalStateException("Forbidden call on a closed session.");

    // Checks if the topic to retrieve is the administration topic:
    if (topicName.equals("#AdminTopic")) {
      try {
        GetAdminTopicReply reply =  
          (GetAdminTopicReply) cnx.syncRequest(new GetAdminTopicRequest());
        if (reply.getId() != null)
          return new Topic(reply.getId());
        else
          throw new JMSException("AdminTopic could not be retrieved.");
      }
      catch (JMSException exc) {
        throw exc;
      }
      catch (Exception exc) {
        throw new JMSException("AdminTopic could not be retrieved: " + exc);
      }
    }
    return new Topic(topicName);
  }

  public TemporaryQueue createTemporaryQueue() throws JMSException
  {
    if (closed)
      throw new IllegalStateException("Forbidden call on a closed session.");

    SessCreateTDReply reply =
      (SessCreateTDReply) cnx.syncRequest(new SessCreateTQRequest());
    String tempDest = reply.getAgentId();
    return new TemporaryQueue(tempDest, cnx);
  }

  public TemporaryTopic createTemporaryTopic() throws JMSException
  {
    if (closed)
      throw new IllegalStateException("Forbidden call on a closed session.");

    SessCreateTDReply reply =
      (SessCreateTDReply) cnx.syncRequest(new SessCreateTTRequest());
    String tempDest = reply.getAgentId();
    return new TemporaryTopic(tempDest, cnx);
  }

  public synchronized void run()
  {
    int load = repliesIn.size();
    com.scalagent.kjoram.messages.Message momMsg;
    String msgId;
    String targetName = connectionConsumer.targetName;
    boolean queueMode = connectionConsumer.queueMode;

    if (JoramTracing.dbgClient)
      JoramTracing.log(JoramTracing.DEBUG, "-- " + this
                       + ": loaded with " + load
                       + " message(s) and started.");
    try { 
      // Processing the current number of messages in the queue:
      for (int i = 0; i < load; i++) {
        momMsg = (com.scalagent.kjoram.messages.Message) repliesIn.pop();
        msgId = momMsg.getIdentifier();

        // If no message listener has been set for the session, denying the
        // processed message:
        if (messageListener == null) {
          JoramTracing.log(JoramTracing.ERROR,this + ": an"
                           + " asynchronous delivery arrived for"
                           + " a non existing session listener:"
                           + " denying the message.");
          
          if (queueMode) 
            cnx.syncRequest(new ConsumerDenyRequest(targetName, msgId, true));
          else
            cnx.asyncRequest(new ConsumerDenyRequest(targetName, msgId,false));
        }
        // Else:
        else {
          // Preparing ack for manual sessions:
          if (! autoAck)
            prepareAck(targetName, msgId, queueMode);
  
          // Passing the current message:
          try {
            messageListener.onMessage(Message.wrapMomMessage(this, momMsg));
  
            // Auto ack: acknowledging the message:
            if (autoAck)
              cnx.asyncRequest(new ConsumerAckRequest(targetName, msgId,
                                                      queueMode));
          }
          // Catching a JMSException means that the building of the Joram
          // message went wrong: denying the message:
          catch (JMSException jE) {
            JoramTracing.log(JoramTracing.ERROR, this
                             + ": error while processing the"
                             + " received message: " + jE);
            if (queueMode)
              cnx.syncRequest(new ConsumerDenyRequest(targetName, msgId,
                                                      queueMode));
            else
              cnx.asyncRequest(new ConsumerDenyRequest(targetName, msgId,
                                                       queueMode));
          }
          // Catching a RuntimeException means that the client onMessage() code
          // is incorrect; denying the message if needed:
          catch (RuntimeException rE) {
            JoramTracing.log(JoramTracing.ERROR,this
                             + ": RuntimeException thrown"
                             + " by the listener: " + rE);
            if (autoAck && queueMode)
              cnx.syncRequest(new ConsumerDenyRequest(targetName, msgId,
                                                      queueMode));
            else if (autoAck && ! queueMode)
              cnx.asyncRequest(new ConsumerDenyRequest(targetName, msgId,
                                                       queueMode));
          }
        }
      }
    }
    catch (JMSException e) {}
  }

  public void commit() throws JMSException
  {
    if (closed)
      throw new IllegalStateException("Forbidden call on a closed session.");

    if (! transacted)
      throw new IllegalStateException("Can't commit a non transacted"
                                      + " session.");

    if (JoramTracing.dbgClient)
      JoramTracing.log(JoramTracing.DEBUG, "--- " + this
                       + ": committing...");

    // If the transaction was scheduled: cancelling.
    if (scheduled) {
      closingTask.cancel();
      scheduled = false;
    }

    // Sending client messages:
    try {
      Enumeration dests = sendings.keys();
      String dest;
      ProducerMessages pM;
      while (dests.hasMoreElements()) {
        dest = (String) dests.nextElement();
        pM = (ProducerMessages) sendings.remove(dest);
        cnx.syncRequest(pM);
      }
      // Acknowledging the received messages:
      acknowledge();

      if (JoramTracing.dbgClient)
        JoramTracing.log(JoramTracing.DEBUG, this + ": committed.");
    }
    // Catching an exception if the sendings or acknowledgement went wrong:
    catch (JMSException jE) {
      TransactionRolledBackException tE = 
        new TransactionRolledBackException("A JMSException was thrown during"
                                           + " the commit.");
      tE.setLinkedException(jE);

      JoramTracing.log(JoramTracing.ERROR, "Exception: " + tE);
      
      rollback();
      throw tE;
    }
  }

  public void rollback() throws JMSException
  {
    if (closed)
      throw new IllegalStateException("Forbidden call on a closed session.");

    if (! transacted)
      throw new IllegalStateException("Can't rollback a non transacted"
                                      + " session.");

    if (JoramTracing.dbgClient)
      JoramTracing.log(JoramTracing.DEBUG, "--- " + this
                       + ": rolling back...");

    // If the transaction was scheduled: cancelling.
    if (scheduled) {
      closingTask.cancel();
      scheduled = false;
    }

    // Denying the received messages:
    deny();
    // Deleting the produced messages:
    sendings.clear();

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

  public void recover() throws JMSException
  {
    if (transacted)
      throw new IllegalStateException("Can't recover a transacted session.");

    if (JoramTracing.dbgClient)
      JoramTracing.log(JoramTracing.DEBUG, "--- " + this
                       + " recovering...");

    // Stopping the session, denying the received messages:
    stop();
    deny();
    // Re-starting the session:
    start();

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


  public void unsubscribe(String name) throws JMSException
  {
    MessageConsumer cons;
    for (int i = 0; i < consumers.size(); i++) {
      cons = (MessageConsumer) consumers.elementAt(i);
      if (! cons.queueMode && cons.targetName.equals(name))
        throw new JMSException("Can't delete durable subscription " + name
                               + " as long as an active subscriber exists.");
    }
    cnx.syncRequest(new ConsumerUnsubRequest(name));
  }

  public synchronized void close() throws JMSException
  {
    // Ignoring the call if the session is already closed:
    if (closed)
      return;

    if (JoramTracing.dbgClient)
      JoramTracing.log(JoramTracing.DEBUG, "--- " + this
                       + ": closing..."); 

    // Finishing the timer, if any:
    if (consumersTimer != null)
      consumersTimer.cancel();

    // Emptying the current pending deliveries:
    try {
      repliesIn.stop();
    }
    catch (InterruptedException iE) {}

    // Stopping the session:
    stop();

    // Denying the non acknowledged messages:
    if (transacted)
      rollback();
    else
      deny();
      
    // Closing the session's resources:
    while (! browsers.isEmpty())
      ((QueueBrowser) browsers.elementAt(0)).close();
    while (! consumers.isEmpty())
      ((MessageConsumer) consumers.elementAt(0)).close();
    while (! producers.isEmpty())
      ((MessageProducer) producers.elementAt(0)).close();

    cnx.sessions.removeElement(this);

    closed = true;

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

  synchronized void schedule(TimerTask task, long timer)
  {
    if (consumersTimer == null)
      consumersTimer = new com.scalagent.kjoram.util.Timer();

    try {
      consumersTimer.schedule(task, timer);
    }
    catch (Exception exc) {}
  }
  
  void start() throws IllegalStateException
  {
    if (closed)
      throw new IllegalStateException("Forbidden call on a closed session.");

    if (JoramTracing.dbgClient)
      JoramTracing.log(JoramTracing.DEBUG, "--- " + this
                       + ": starting...");

    repliesIn.start();

    // Starting the daemon if needed:
    if (! started && msgListeners > 0) {
      daemon = new SessionDaemon(this);
      daemon.setDaemon(false);
      daemon.start();
    }
    started = true;

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

  void stop()
  {
    // Ignoring the call if the session is already stopped:
    if (! started)
      return;

    if (JoramTracing.dbgClient)
      JoramTracing.log(JoramTracing.DEBUG, "--- " + this
                       + ": stopping...");

    // Stopping the daemon if needed:
    if (daemon != null) {
      daemon.stop();
      daemon = null;
    }
    // Synchronizing the stop() with the consumers:
    if (consumers != null) {
      MessageConsumer consumer; 
      for (int i = 0; i < consumers.size(); i++) {
        consumer = (MessageConsumer) consumers.elementAt(i);
        consumer.syncro();
      }
    }

    started = false;

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

  void prepareSend(Destination dest, com.scalagent.kjoram.messages.Message msg)
  {
    // If the transaction was scheduled, cancelling:
    if (scheduled)
      closingTask.cancel();

    ProducerMessages pM = (ProducerMessages) sendings.get(dest.getName());
    if (pM == null) {
      pM = new ProducerMessages(dest.getName());
      sendings.put(dest.getName(), pM);
    }
    pM.addMessage(msg);

    // If the transaction was scheduled, re-scheduling it:
    if (scheduled)
      cnx.schedule(closingTask);
  }

  void prepareAck(String name, String id, boolean queueMode)
  {
    // If the transaction was scheduled, cancelling:
    if (scheduled)
      closingTask.cancel();

    MessageAcks acks = (MessageAcks) deliveries.get(name);
    if (acks == null) {
      acks = new MessageAcks(queueMode);
      deliveries.put(name, acks);
    }
    acks.addId(id);

    // If the transaction must be scheduled, scheduling it:
    if (closingTask != null) {
      scheduled = true;
      cnx.schedule(closingTask);
    }
  }

  void acknowledge() throws IllegalStateException
  { 
    String target;
    MessageAcks acks;

    Enumeration targets = deliveries.keys();
    while (targets.hasMoreElements()) {
      target = (String) targets.nextElement();
      acks = (MessageAcks) deliveries.remove(target);
      cnx.asyncRequest(new SessAckRequest(target, acks.getIds(),
                                          acks.getQueueMode()));
    }
  }

  void deny()
  {
    try {
      String target;
      MessageAcks acks;
      SessDenyRequest deny;

      Enumeration targets = deliveries.keys();
      while (targets.hasMoreElements()) {
        target = (String) targets.nextElement();
        acks = (MessageAcks) deliveries.remove(target);
        deny = new SessDenyRequest(target, acks.getIds(), acks.getQueueMode());
        if (acks.getQueueMode())
          cnx.syncRequest(deny);
        else
          cnx.asyncRequest(deny);
      }
    }
    catch (JMSException jE) {}
  }

  void distribute(AbstractJmsReply asyncReply)
  {
    // Getting the message:
    ConsumerMessages reply = (ConsumerMessages) asyncReply;

    // Getting the consumer:
    MessageConsumer cons = null;
    if (reply.getQueueMode()) {
      cons =
        (MessageConsumer) cnx.requestsTable.remove(reply.getKey());
    }
    else
      cons = (MessageConsumer) cnx.requestsTable.get(reply.getKey());

    // Passing the message(s) to the consumer:
    if (cons != null) {
      Vector msgs = reply.getMessages();
      for (int i = 0; i < msgs.size(); i++) 
        cons.onMessage((com.scalagent.kjoram.messages.Message) msgs.elementAt(i));
    }
    // The target consumer of the received message may be null if it has
    // been closed without having stopped the connection: denying the
    // deliveries.
    else {
      if (JoramTracing.dbgClient)
        JoramTracing.log(JoramTracing.WARN, this + ": an asynchronous"
                         + " delivery arrived for an improperly"
                         + " closed consumer: denying the"
                         + " messages.");

      Vector msgs = reply.getMessages();
      com.scalagent.kjoram.messages.Message msg;
      Vector ids = new Vector();
      for (int i = 0; i < msgs.size(); i++) {
        msg = (com.scalagent.kjoram.messages.Message) msgs.elementAt(i);
        ids.addElement(msg.getIdentifier());
      }
  
      if (ids.isEmpty())
        return;
  
      try { 
        cnx.asyncRequest(new SessDenyRequest(reply.comesFrom(), ids,
                                             reply.getQueueMode(), true));
      }
      catch (JMSException jE) {}
    }
  }

  private class SessionCloseTask extends TimerTask
  {
    public void run()
    {
      try {
        if (JoramTracing.dbgClient)
          JoramTracing.log(JoramTracing.WARN, "Session closed "
                           + "because of pending transaction");
        close();
      }
      catch (Exception e) {}
    }
  }
}

---