com/scalagent/kjoram/MessageProducer.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.IllegalStateException;
import com.scalagent.kjoram.excepts.*;

public class MessageProducer
{
  private int deliveryMode = DeliveryMode.PERSISTENT;
  private int priority = 4;
  private long timeToLive = 0;
  private boolean messageIDDisabled = false;
  private boolean timestampDisabled = false;
  private boolean identified = true;

  protected boolean closed = false;
  protected Session sess;
  protected Destination dest = null;
  

  MessageProducer(Session sess, Destination dest) throws JMSException
    {
      this.sess = sess;
      this.dest = dest;

      if (dest == null)
        identified = false;

      sess.producers.addElement(this);

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

  public void setDisableMessageID(boolean value) throws JMSException
    {
      if (closed)
        throw new IllegalStateException("Forbidden call on a closed producer.");
    }

  public void setDeliveryMode(int deliveryMode) throws JMSException
    {
      if (closed)
        throw new IllegalStateException("Forbidden call on a closed producer.");

      if (deliveryMode != DeliveryMode.PERSISTENT
          && deliveryMode != DeliveryMode.NON_PERSISTENT)
        throw new JMSException("Can't set invalid delivery mode.");

      this.deliveryMode = deliveryMode;
    }

  public void setPriority(int priority) throws JMSException
    {
      if (closed)
        throw new IllegalStateException("Forbidden call on a closed producer.");

      if (priority < 0 || priority > 9)
        throw new JMSException("Can't set invalid priority value.");

      this.priority = priority;
    }

  public void setTimeToLive(long timeToLive) throws JMSException
    {
      if (closed)
        throw new IllegalStateException("Forbidden call on a closed producer.");

      this.timeToLive = timeToLive;
    }

  public void setDisableMessageTimestamp(boolean value) throws JMSException
    {
      if (closed)
        throw new IllegalStateException("Forbidden call on a closed producer.");

      this.timestampDisabled = value;
    }

  public Destination getDestination() throws JMSException
    {
      if (closed)
        throw new IllegalStateException("Forbidden call on a closed producer.");

      return dest;
    }

  public boolean getDisableMessageID() throws JMSException
    {
      if (closed)
        throw new IllegalStateException("Forbidden call on a closed producer.");

      return messageIDDisabled;
    }

  public int getDeliveryMode() throws JMSException
    {
      if (closed)
        throw new IllegalStateException("Forbidden call on a closed producer.");

      return deliveryMode;
    }

  public int getPriority() throws JMSException
    {
      if (closed)
        throw new IllegalStateException("Forbidden call on a closed producer.");

      return priority;
    }

  public long getTimeToLive() throws JMSException
    {
      if (closed)
        throw new IllegalStateException("Forbidden call on a closed producer.");

      return timeToLive;
    }

  public boolean getDisableMessageTimestamp() throws JMSException
    {
      if (closed)
        throw new IllegalStateException("Forbidden call on a closed producer.");

      return timestampDisabled;
    }


  public void send(Message message) throws JMSException
    {
      if (! identified)
        throw new RuntimeException("Can't send message to"
                                   + " an unidentified"
                                   + " destination.");
      // Actually producing it:
      doSend(dest, message, deliveryMode, priority, timeToLive);
    }

  public void send(Message message, int deliveryMode,
                   int priority, long timeToLive) throws JMSException
    {
      if (! identified)
        throw new RuntimeException("Can't send message to"
                                   + " an unidentified"
                                   + " destination.");
      // Actually producing it:
      doSend(dest, message, deliveryMode, priority, timeToLive);
    }

  public void send(Destination dest,
                   Message message) throws JMSException
    {
      if (identified)
        throw new RuntimeException("An unidentified message"
                                   + " producer can't use this"
                                   + " identified message"
                                   + " producer.");
      if (dest == null)
        throw new RuntimeException("Can't send message to"
                                   + " an unidentified"
                                   + " destination.");
      
      doSend((Destination) dest, message, deliveryMode, priority, timeToLive);
    }

  public void send(Destination dest, Message message,
                   int deliveryMode, int priority,
                   long timeToLive) throws JMSException
    {
      if (identified)
        throw new RuntimeException("An unidentified message"
                                   + " producer can't use this"
                                   + " identified message"
                                   + " producer.");
      if (dest == null)
        throw new RuntimeException("Can't send message to"
                                   + " an unidentified"
                                   + " destination.");

      doSend((Destination) dest, message, deliveryMode, priority, timeToLive);
    }

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

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

      sess.producers.removeElement(this);
      closed = true;

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

    }

  private void doSend(Destination dest, Message message,
                      int deliveryMode, int priority,
                      long timeToLive) throws JMSException
    {
      if (closed)
        throw new IllegalStateException("Forbidden call on a closed producer.");

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

      // Updating the message property fields:
      String msgID = sess.cnx.nextMessageId();
      message.setJMSMessageID(msgID);
      message.setJMSDeliveryMode(deliveryMode);
      message.setJMSDestination(dest);
      if (timeToLive == 0)
        message.setJMSExpiration(0);
      else
        message.setJMSExpiration(System.currentTimeMillis() + timeToLive);
      message.setJMSPriority(priority);
      if (! timestampDisabled)
        message.setJMSTimestamp(System.currentTimeMillis());

      com.scalagent.kjoram.messages.Message momMsg = null;
      // If the message to send is a proprietary one, getting the MOM message
      // it wraps:
      if (message instanceof Message)
        momMsg = ((Message) message).getMomMessage();

      // If the message to send is a non proprietary JMS message, building
      // a proprietary message and then getting the MOM message it wraps:
      else if (message instanceof Message) {
        try {
          Message joramMessage = Message.convertJMSMessage(message);
          momMsg = joramMessage.getMomMessage();
        }
        catch (JMSException jE) {
          MessageFormatException mE = new MessageFormatException("Message to"
                                                                 + " send is"
                                                                 + " invalid.");
          mE.setLinkedException(jE);
          throw mE;
        }
      }
      else {
        MessageFormatException mE = new MessageFormatException("Message to"
                                                               + " send is"
                                                               + " invalid.");
        throw mE;
      }

      // If the session is transacted, keeping the request for later delivery:
      if (sess.transacted) {
        if (JoramTracing.dbgClient)
          JoramTracing.log(JoramTracing.DEBUG, "Buffering the message.");

        sess.prepareSend(dest,
                         (com.scalagent.kjoram.messages.Message) momMsg.clone());
      }
      // If not, building a new request and sending it:
      else {
      ProducerMessages pM = new ProducerMessages(dest.getName(), momMsg);

      if (JoramTracing.dbgClient)
        JoramTracing.log(JoramTracing.DEBUG, "Sending " + momMsg);
      
        sess.cnx.syncRequest(pM);
      }
    }
}

---