message.java

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

/*
 * JORAM: Java(TM) Open Reliable Asynchronous Messaging
 * Copyright (C) 2001 - 2006 ScalAgent Distributed Technologies
 * Copyright (C) 1996 - 2000 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): ScalAgent Distributed Technologies
 */
package org.objectweb.joram.shared.messages;

import org.objectweb.joram.shared.excepts.*;
import org.objectweb.joram.mom.util.MessagePersistenceModule;
import org.objectweb.joram.shared.messages.MessageTracing;
import org.objectweb.util.monolog.api.BasicLevel;

import java.io.*;
import java.util.*;

/** 
 * The <code>Message</code> class actually provides the transport facility
 * for the data exchanged during MOM operations.
 * <p>
 * A message may either carry a String, or a serializable object, or an
 * hashtable, or bytes, even nothing. It is charaterized by properties and
 * "header" fields.
 */
public class Message implements Cloneable, Serializable {
  transient public boolean pin = true;
  transient private String saveName = null;
  transient public boolean toBeUpdated = false;
  transient protected MessagePersistent messagePersistent = null;
  transient public boolean noBody = false;

  /** The message type (SIMPLE, TEXT, OBJECT, MAP, STREAM, BYTES). */
  transient int type;

  /** <code>true</code> if the message must be persisted. **/
  transient boolean persistent = true;

  /** The message identifier. */
  transient String id = null;
  /** The message priority (from 0 to 9, 9 being the highest). */
  transient int priority = 4;
  /** The message expiration time (0 for infinite time-to-live). */
  transient long expiration = 0;
  /** The message time stamp. */
  transient long timestamp;
  /** The message destination identifier. */
  transient String toId = null;
  /** The message destination type. */
  transient String toType;
  /** The correlation identifier field. */
  transient String correlationId = null;
  /** The reply to destination identifier. */
  transient String replyToId = null;
  /** <code>true</code> if the "reply to" destination is a queue. */
  transient String replyToType;

  /**
   * Table holding header fields that may be required by particular
   * clients (such as JMS clients).
   */
  transient Hashtable optionalHeader = null;

  /** <code>true</code> if the body is read-only. */
  transient boolean bodyRO = false;
  transient protected MessageBody body = null;

  /** The message properties table. */
  transient Hashtable properties = null;
  /** <code>true</code> if the properties are read-only. */
  transient boolean propertiesRO = false;

  /** Arrival position of this message on its queue or proxy. */
  transient public long order;
  /** The number of delivery attempts for this message. */
  transient public int deliveryCount = 0;
  /**
   * <code>true</code> if the message has been denied at least once by a
   * consumer.
   */
  transient public boolean denied = false;
  
  /** <code>true</code> if the message target destination is deleted. */
  transient public boolean deletedDest = false;
  /** <code>true</code> if the message expired. */
  transient public boolean expired = false;
  /** <code>true</code> if the message could not be written on the dest. */
  transient public boolean notWriteable = false;
  /** <code>true</code> if the message is considered as undeliverable. */
  transient public boolean undeliverable = false;

  /**
   * The number of acknowledgements a message still expects from its 
   * subscribers before having been fully consumed by them (field used
   * by JMS proxies).
   */
  public transient int acksCounter;
  /**
   * The number of acknowledgements a message still expects from its 
   * durable subscribers before having been fully consumed by them (field used
   * by JMS proxies).
   */
  public transient int durableAcksCounter;

  /**
   * Constructs a <code>Message</code> instance.
   */
  public Message() {
    this.type = MessageType.SIMPLE;
    body = new MessageBody();

    if (MessageTracing.dbgMessage.isLoggable(BasicLevel.DEBUG))
      MessageTracing.dbgMessage.log(BasicLevel.DEBUG,
                                    "Message <init>");
  }

  public static Message create() {
    if (MessageTracing.dbgMessage.isLoggable(BasicLevel.DEBUG))
      MessageTracing.dbgMessage.log(BasicLevel.DEBUG,
                                    "Message.create()");
    Message message;
    try {
      String momMessageClass = 
        System.getProperty("MomMessageClass", 
                           "org.objectweb.joram.shared.messages.Message");
      message =  (Message) Class.forName(momMessageClass).newInstance();
    } catch (Exception exc) {
      MessageTracing.dbgMessage.log(BasicLevel.ERROR,
                                    "Message.create() : " +
                                    "return new org.objectweb.joram.shared.messages.Message",
                                    exc);
      message =  new Message();
    }
      message.setPin(true);
      return message;
  }

  public void setPin(boolean pin) {
    this.pin = pin;
  }

  public boolean isPin() {
    return pin;
  }

  public void setSaveName(String saveName) {
    this.saveName = saveName;
  }

  public String getSaveName() {
    return saveName;
  }

  protected MessageBody getMessageBody() {
    return body;
  }

  public void setMessageBody(MessageBody body) {
    this.body = body;
  }

  /** get bytes body. */
  protected byte[] getBodyBytes() {
    return getMessageBody().getBodyBytes();
  }
  
  /** get map body. */
  protected HashMap getBodyMap() {
    return getMessageBody().getBodyMap();
  }
  
  /** get text body. */
  protected String getBodyText() {
    return getMessageBody().getBodyText();
  }
  
  /** set bytes body. */
  protected void setBodyBytes(byte[] bytes) {
    MessageBody body = getMessageBody();
    body.setType(type);
    body.setBodyBytes(bytes);
  }
  
  /** set map body. */
  protected void setBodyMap(HashMap map) {
    MessageBody body = getMessageBody();
    body.setType(type);
    body.setBodyMap(map);
  }

  /** set text body. */
  protected void setBodyText(String text) {
    MessageBody body = getMessageBody();
    body.setType(type);
    body.setBodyText(text);
  }

  /** Sets the message identifier. */ 
  public void setIdentifier(String id) {
    this.id = id;
  }

  /** Sets the message persistence mode. */
  public void setPersistent(boolean persistent) {
    this.persistent = persistent;
  }

  /**
   * Sets the message priority.
   *
   * @param priority  Priority value: 0 the lowest, 9 the highest, 4 normal.
   */ 
  public void setPriority(int priority) {
    if (priority >= 0 && priority <= 9)
      this.priority = priority;
  }

  /**
   * Sets the message expiration.
   *
   * @param expiration	The expiration time.
   */
  public void setExpiration(long expiration) {
    if (expiration >= 0)
      this.expiration = expiration;
  }

  /** Sets the message time stamp. */
  public void setTimestamp(long timestamp) {
    this.timestamp = timestamp;
  }

  /**
   * Sets the message destination.
   *
   * @param id  The destination identifier.
   * @param type The type of the destination.
   */
  public void setDestination(String id, String type) {
    this.toId = id;
    this.toType = type;
  }

  /** Sets the message correlation identifier. */
  public void setCorrelationId(String correlationId) {
    this.correlationId = correlationId;
  }

  /**
   * Sets the destination to which a reply should be sent.
   *
   * @param id  The destination identifier.
   * @param type The destination type.
   */
  public void setReplyTo(String id, String type) {
    this.replyToId = id;
    this.replyToType = type;
  }

  /**
   * Sets an optional header field value.
   *
   * @param name  The header field name.
   * @param value  The corresponding value.
   */
  public void setOptionalHeader(String name, Object value) {
    if (name == null || name.equals(""))
      throw new IllegalArgumentException("Invalid header name: " + name);

    if (value == null)
      return;

    if (optionalHeader == null)
      optionalHeader = new Hashtable();

    optionalHeader.put(name, value);
  }

  /**
   *  Copies all of the mappings from the optionalHeader of this message to
   * the specified hashtable. These mappings will replace any mappings that
   * this Hashtable had for any of the keys currently in the optional header.
   */
  public void getOptionalHeader(Hashtable h) {
    if (optionalHeader != null)
      h.putAll(optionalHeader);
  }

  /** Returns the message type. */
  public int getType() {
    return type;
  }

  /** Returns the message identifier. */
  public String getIdentifier() {
    return id;
  }

  /** Returns <code>true</code> if the message is persistent. */
  public boolean getPersistent() {
    return persistent;
  }

  /** Returns the message priority. */
  public int getPriority() {
    return priority;
  }

  /** Returns the message expiration time. */
  public long getExpiration() {
    return expiration;
  }
  
  /** Returns the message time stamp. */
  public long getTimestamp() {
    return timestamp;
  }

  /** Returns the message destination identifier. */
  public final String getDestinationId() {
    return toId;
  }

  /** Returns <code>true</code> if the destination is a queue. */
  public final String toType() {
    return toType;
  }

  /** Returns the message correlation identifier. */
  public final String getCorrelationId() {
    return correlationId;
  }

  /** Returns the destination id the reply should be sent to. */
  public final String getReplyToId() {
    return replyToId;
  }

  /** Returns <code>true</code> if the reply to destination is a queue. */
  public final String replyToType() {
    return replyToType;
  }

  /**
   * Returns an optional header field value.
   *
   * @param name  The header field name.
   */
  public Object getOptionalHeader(String name) {
    if (optionalHeader == null)
      return null;

    return optionalHeader.get(name);
  }

  /**
   * Sets a property as a boolean value.
   *
   * @param name  The property name.
   * @param value  The property value.
   *
   * @exception MessageROException  If the message properties are read-only.
   */
  public void setBooleanProperty(String name, boolean value) throws MessageROException {
    setProperty(name, new Boolean(value));
  }

  /**
   * Sets a property as a byte value.
   *
   * @param name  The property name.
   * @param value  The property value.
   *
   * @exception MessageROException  If the message properties are read-only.
   */
  public void setByteProperty(String name, byte value) throws MessageROException {
    setProperty(name, new Byte(value));
  }