session.java

一个类似于openJMS分布在ObjectWeb之下的JMS消息中间件。

  public synchronized javax.jms.Message createMessage() 
    throws JMSException {
    checkClosed();
    return new Message();
  }

  /**
   * API method.
   *
   * @exception IllegalStateException  If the session is closed.
   */
  public synchronized javax.jms.TextMessage createTextMessage() 
    throws JMSException {
    checkClosed();
    return new TextMessage();
  }

  /**
   * API method.
   *
   * @exception IllegalStateException  If the session is closed.
   */
  public synchronized javax.jms.TextMessage createTextMessage(String text)
    throws JMSException {
    checkClosed();
    TextMessage message =  new TextMessage();
    message.setText(text);
    return message;
  }
  
  /**
   * API method.
   *
   * @exception IllegalStateException  If the session is closed.
   */
  public synchronized javax.jms.BytesMessage createBytesMessage()
    throws JMSException {
    checkClosed();
    return new BytesMessage();
  }

  /**
   * API method.
   *
   * @exception IllegalStateException  If the session is closed.
   */
  public synchronized javax.jms.MapMessage createMapMessage()
    throws JMSException {
    checkClosed();
    return new MapMessage();
  }

  /**
   * API method.
   *
   * @exception IllegalStateException  If the session is closed.
   */
  public synchronized javax.jms.ObjectMessage createObjectMessage()
    throws JMSException {
    checkClosed();
    return new ObjectMessage();
  }

  /**
   * API method.
   *
   * @exception IllegalStateException  If the session is closed.
   */
  public synchronized javax.jms.ObjectMessage createObjectMessage(
    java.io.Serializable obj)
    throws JMSException {
    checkClosed();
    ObjectMessage message = new ObjectMessage(); 
    message.setObject(obj);
    return message;
  }

  /**
   * API method.
   *
   * @exception IllegalStateException  If the session is closed.
   */
  public synchronized javax.jms.StreamMessage createStreamMessage()
    throws JMSException {
    checkClosed();
    return new StreamMessage();
  }

  /**
   * API method
   *
   * @exception IllegalStateException  If the session is closed.
   */
  public synchronized javax.jms.QueueBrowser
      createBrowser(javax.jms.Queue queue, 
                    String selector)
    throws JMSException {
    checkClosed();
    checkThreadOfControl();
    QueueBrowser qb = new QueueBrowser(this, (Queue) queue, selector);
    browsers.addElement(qb);
    return qb;
  }

  /**
   * API method
   *
   * @exception IllegalStateException  If the session is closed.
   */
  public synchronized javax.jms.QueueBrowser 
      createBrowser(javax.jms.Queue queue)
    throws JMSException {
    checkClosed();
    checkThreadOfControl();
    QueueBrowser qb =  new QueueBrowser(this, (Queue) queue, null);
    browsers.addElement(qb);
    return qb;
  }

  /**
   * API method.
   *
   * @exception IllegalStateException  If the session is closed or if the 
   *              connection is broken.
   * @exception JMSException  If the creation fails for any other reason.
   */
  public synchronized javax.jms.MessageProducer createProducer(
    javax.jms.Destination dest)
    throws JMSException {
    checkClosed();
    checkThreadOfControl();
    MessageProducer mp = new MessageProducer(
      this, 
      (Destination) dest);
    addProducer(mp);
    return mp;
  }

  /**
   * API method.
   *
   * @exception IllegalStateException  If the session is closed or if the
   *              connection is broken.
   * @exception JMSException  If the creation fails for any other reason.
   */
  public synchronized javax.jms.MessageConsumer
      createConsumer(javax.jms.Destination dest, 
                     String selector,
                     boolean noLocal) 
    throws JMSException {
    checkClosed();
    checkThreadOfControl();
    MessageConsumer mc = new MessageConsumer(
      this, (Destination) dest, 
      selector, null,
      noLocal);
    addConsumer(mc);
    return mc;
  }

  /**
   * API method.
   *
   * @exception IllegalStateException  If the session is closed or if the
   *              connection is broken.
   * @exception JMSException  If the creation fails for any other reason.
   */
  public synchronized javax.jms.MessageConsumer
      createConsumer(javax.jms.Destination dest, 
                     String selector)
    throws JMSException {
    checkClosed();
    checkThreadOfControl();
    MessageConsumer mc = new MessageConsumer(
      this, (Destination) dest, selector);
    addConsumer(mc);
    return mc;
  }

  /**
   * API method.
   *
   * @exception IllegalStateException  If the session is closed or if the
   *              connection is broken.
   * @exception JMSException  If the creation fails for any other reason.
   */
  public synchronized javax.jms.MessageConsumer 
      createConsumer(javax.jms.Destination dest)
    throws JMSException {
    checkClosed();
    checkThreadOfControl();
    MessageConsumer mc = new MessageConsumer(
      this, (Destination) dest, null);
    addConsumer(mc);
    return mc;
  }

  /**
   * API method.
   *
   * @exception IllegalStateException  If the session is closed or if the 
   *              connection is broken.
   * @exception JMSException  If the creation fails for any other reason.
   */
  public synchronized javax.jms.TopicSubscriber
      createDurableSubscriber(javax.jms.Topic topic, 
                              String name,
                              String selector,
                              boolean noLocal) 
    throws JMSException {
    if (logger.isLoggable(BasicLevel.DEBUG))
      logger.log(
        BasicLevel.DEBUG, 
        "Session.createDurableSubscriber(" + 
        topic + ',' + name + ',' + 
        selector + ',' + noLocal + ')');
    checkClosed();
    checkThreadOfControl();
    TopicSubscriber ts = new TopicSubscriber(
      this, (Topic) topic, name, selector, noLocal);
    addConsumer(ts);
    return ts;
  }

  /**
   * API method.
   *
   * @exception IllegalStateException  If the session is closed or if the 
   *              connection is broken.
   * @exception JMSException  If the creation fails for any other reason.
   */
  public synchronized javax.jms.TopicSubscriber
      createDurableSubscriber(javax.jms.Topic topic, 
                              String name)
    throws JMSException {
    if (logger.isLoggable(BasicLevel.DEBUG))
      logger.log(
        BasicLevel.DEBUG, 
        "Session.createDurableSubscriber(" + 
        topic + ',' + name + ')');
    checkClosed();
    checkThreadOfControl();
    TopicSubscriber ts = new TopicSubscriber(
      this, (Topic) topic, name, null, false);
    addConsumer(ts);
    return ts;
  }

  /**
   * API method.
   *
   * @exception IllegalStateException  If the session is closed.
   */
  public synchronized javax.jms.Queue createQueue(
    String queueName) 
    throws JMSException {
    checkClosed();
    return new Queue(queueName);
  }

  /**
   * API method.
   *
   * @exception IllegalStateException  If the session is closed.
   * @exception JMSException  If the topic creation failed.
   */
  public synchronized javax.jms.Topic createTopic(
    String topicName) 
    throws JMSException {
    checkClosed();
    checkThreadOfControl();

    // Checks if the topic to retrieve is the administration topic:
    if (topicName.equals("#AdminTopic")) {
      try {
        GetAdminTopicReply reply =  
          (GetAdminTopicReply) requestor.request(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);
  }

  /**
   * API method.
   *
   * @exception IllegalStateException  If the session is closed or if the
   *              connection is broken.
   * @exception JMSException  If the request fails for any other reason.
   */
  public synchronized javax.jms.TemporaryQueue createTemporaryQueue() 
    throws JMSException {
    checkClosed();
    checkThreadOfControl();

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

  /**
   * API method.
   *
   * @exception IllegalStateException  If the session is closed or if the
   *              connection is broken.
   * @exception JMSException  If the request fails for any other reason.
   */
  public synchronized javax.jms.TemporaryTopic createTemporaryTopic() 
    throws JMSException {
    checkClosed();
    checkThreadOfControl();

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

  /** API method. */
  public synchronized void run() {
    int load = repliesIn.size();

    if (logger.isLoggable(BasicLevel.DEBUG))
      logger.log(BasicLevel.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++) {
        org.objectweb.joram.shared.messages.Message momMsg = 
          (org.objectweb.joram.shared.messages.Message) repliesIn.pop();
        String msgId = momMsg.getIdentifier();
        
        onMessage(momMsg, messageConsumerListener);
      }
    } catch (JMSException exc) {
      if (logger.isLoggable(BasicLevel.ERROR))
        logger.log(BasicLevel.ERROR, "", exc);
    }
  }
  
  /**
   * Called by MultiSessionConsumer
   * ASF mode
   */
  void setMessageConsumerListener(MessageConsumerListener mcl) {
    messageConsumerListener = mcl;
  }
      
  /**
   * API method.
   *
   * @exception IllegalStateException  If the session is closed, or not
   *              transacted, or if the connection is broken.
   */
  public synchronized void commit() throws JMSException {
    if (logger.isLoggable(BasicLevel.DEBUG))
      logger.log(
        BasicLevel.DEBUG,
        "Session.commit()");

    checkClosed();
    checkThreadOfControl();

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

    if (logger.isLoggable(BasicLevel.DEBUG))
      logger.log(BasicLevel.DEBUG, "--- " + this
                                 + ": committing...");

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

    // Sending client messages:
    try {
      CommitRequest commitReq= new CommitRequest();
      
      Enumeration producerMessages = sendings.elements();
      while (producerMessages.hasMoreElements()) {
        ProducerMessages pM = 
          (ProducerMessages) producerMessages.nextElement();
        commitReq.addProducerMessages(pM);
      }
      sendings.clear();
      
      // Acknowledging the received messages:
      Enumeration targets = deliveries.keys();
      while (targets.hasMoreElements()) {
        String target = (String) targets.nextElement();
        MessageAcks acks = (MessageAcks) deliveries.get(target);
        commitReq.addAckRequest(
          new SessAckRequest(
            target, 
            acks.getIds(),
            acks.getQueueMode()));
      }
      deliveries.clear();
      
      if (asyncSend) {
        // Asynchronous sending
        commitReq.setAsyncSend(true);
        mtpx.sendRequest(commitReq);
      } else {
        requestor.request(commitReq);
      }

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

      if (logger.isLoggable(BasicLevel.ERROR))
        logger.log(BasicLevel.ERROR, "Exception: " + tE);

      rollback();
      throw tE;
    }
  }

  /**
   * API method.
   *
   * @exception IllegalStateException  If the session is closed, or not
   *              transacted.
   */
  public synchronized void rollback() throws JMSException {
    if (logger.isLoggable(BasicLevel.DEBUG))
      logger.log(
        BasicLevel.DEBUG,
        "Session.rollback()");

    checkClosed();
    checkThreadOfControl();

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

    if (logger.isLoggable(BasicLevel.DEBUG))
      logger.log(BasicLevel.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 (logger.isLoggable(BasicLevel.DEBUG))
      logger.log(BasicLevel.DEBUG, this + ": rolled back.");
  }

  /** 
   * API method.
   *
   * @exception IllegalStateException  If the session is closed, or transacted.
   */
  public synchronized void recover() throws JMSException {
    if (logger.isLoggable(BasicLevel.DEBUG))
      logger.log(
        BasicLevel.DEBUG,
        "Session.recover()");

    checkClosed();
    checkThreadOfControl();

    if (transacted)
      throw new IllegalStateException("Can't recover a transacted session.");
    
    if (logger.isLoggable(BasicLevel.DEBUG))
      logger.log(BasicLevel.DEBUG, "--- " + this
                                 + " recovering...");

    if (daemon != null &&  
        daemon.isCurrentThread()) {
      recover = true;
    } else {
      doRecover();
    }

    if (logger.isLoggable(BasicLevel.DEBUG))
      logger.log(BasicLevel.DEBUG, this + ": recovered.");
  }
  
  private void doRecover() throws JMSException {
    if (logger.isLoggable(BasicLevel.DEBUG))
      logger.log(BasicLevel.DEBUG, "Session.doRecover()");
    deny();
  }

  /**
   * API method.
   *
   * @exception IllegalStateException  If the session is closed or if the 
   *              connection is broken.
   * @exception InvalidDestinationException  If the subscription does not 
   *              exist.
   * @exception JMSException  If the request fails for any other reason.
   */
  public synchronized void unsubscribe(String name)
    throws JMSException {
    if (logger.isLoggable(BasicLevel.DEBUG))
      logger.log(
        BasicLevel.DEBUG,
        "Session.unsubscribe(" + name + ')');

    checkClosed();