message.java

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

  public boolean isValid(long currentTime) {
    if (expiration == 0)
      return true;

    return ((expiration - currentTime) > 0);
  }

  /** Clones the message. */
  public Object clone() {
    try {
      if (MessageTracing.dbgMessage.isLoggable(BasicLevel.DEBUG))
        MessageTracing.dbgMessage.log(BasicLevel.DEBUG,
                                      "Message.clone()");
      Message clone = (Message) super.clone();
      clone.setMessageBody(
        (MessageBody) getMessageBody().clone());

      if (optionalHeader != null) {
        clone.optionalHeader = new Hashtable();
        for (Enumeration e = optionalHeader.keys(); e.hasMoreElements(); ) {
          Object key = e.nextElement();
          clone.optionalHeader.put(key,optionalHeader.get(key));
        }
      }
      if (properties != null) {
        clone.properties = new Hashtable();
        for (Enumeration e = properties.keys(); e.hasMoreElements(); ) {
          Object key = e.nextElement();
          clone.properties.put(key,properties.get(key));
        }
      }
      if (messagePersistent != null) {
        clone.messagePersistent = (MessagePersistent) messagePersistent.clone();
        clone.messagePersistent.message = clone;
      }
      return clone;
    } catch (CloneNotSupportedException cE) {
      return null;
    }
  }

  /**
   * Transforms this message into a table of primitive values that can
   * be vehiculated through the SOAP protocol.
   */
  public Hashtable soapCode() {
    Hashtable h = new Hashtable();

    // Building a hashtable containg the fields values: 
    Hashtable fieldsTb = new Hashtable();

    fieldsTb.put("type", new Integer(type));
    fieldsTb.put("id", id);
    fieldsTb.put("persistent", new Boolean(persistent));
    fieldsTb.put("nobody", new Boolean(noBody));
    fieldsTb.put("priority", new Integer(priority));
    fieldsTb.put("expiration", new Long(expiration));
    fieldsTb.put("timestamp", new Long(timestamp));
    fieldsTb.put("toId", toId);
    fieldsTb.put("toType", toType);
    if (correlationId != null)
      fieldsTb.put("correlationId", correlationId);
    if (replyToId != null) {
      fieldsTb.put("replyToId", replyToId);
      fieldsTb.put("replyToType", replyToType);
    }
    if (getBodyBytes() != null)
      fieldsTb.put("body_bytes", getBodyBytes());
    else if (getBodyMap() != null)
      fieldsTb.put("body_map", new Hashtable(getBodyMap()));
    else if (getBodyText() != null)
      fieldsTb.put("body_text", getBodyText());
    fieldsTb.put("bodyRO", new Boolean(bodyRO));
    fieldsTb.put("propertiesRO", new Boolean(propertiesRO));
    fieldsTb.put("deliveryCount", new Integer(deliveryCount));
    fieldsTb.put("denied", new Boolean(denied));
    fieldsTb.put("deletedDest", new Boolean(deletedDest));
    fieldsTb.put("expired", new Boolean(expired));
    fieldsTb.put("notWriteable", new Boolean(notWriteable));
    fieldsTb.put("undeliverable", new Boolean(undeliverable));

    h.put("fieldsTb",fieldsTb);

    // Adding the hashtable of optional headers:
    if (optionalHeader != null)
      h.put("optionalHeader",optionalHeader);

    // Adding the hashtable of properties:
    if (properties != null)
      h.put("properties",properties);

    return h;
  }

  /** 
   * Transforms a table of primitive values into a <code>Message</code>
   * instance.
   */
  public static Message soapDecode(Hashtable h) {
    if (h == null) return null;

    Hashtable fieldsTb = (Hashtable) h.get("fieldsTb");

    Message msg = Message.create();

    try {
      msg.type = ConversionHelper.toInt(fieldsTb.get("type"));
      msg.id = (String) fieldsTb.get("id");
      msg.persistent = ConversionHelper.toBoolean(fieldsTb.get("persistent"));
      msg.noBody = ConversionHelper.toBoolean(fieldsTb.get("nobody"));
      msg.priority = ConversionHelper.toInt(fieldsTb.get("priority"));
      msg.expiration = ConversionHelper.toLong(fieldsTb.get("expiration"));
      msg.timestamp = ConversionHelper.toLong(fieldsTb.get("timestamp"));
      msg.toId = (String) fieldsTb.get("toId");
      msg.toType = (String)fieldsTb.get("toType");
      msg.correlationId = (String) fieldsTb.get("correlationId");
      msg.replyToId = (String) fieldsTb.get("replyToId");
      if (msg.replyToId != null) {
        msg.replyToType = (String)fieldsTb.get("replyToType");
      }
      msg.setBodyBytes(ConversionHelper.toBytes(fieldsTb.get("body_bytes")));
      Hashtable ht = (Hashtable) fieldsTb.get("body_map");
      if (ht != null) {
        msg.setBodyMap(new HashMap(ht));
      } else {
        msg.setBodyMap((HashMap) null);
      }
      msg.setBodyText((String) fieldsTb.get("body_text"));
      msg.bodyRO = ConversionHelper.toBoolean(fieldsTb.get("bodyRO"));
      msg.propertiesRO =
        ConversionHelper.toBoolean(fieldsTb.get("propertiesRO"));
      msg.deliveryCount =
        ConversionHelper.toInt(fieldsTb.get("deliveryCount"));
      msg.denied = ConversionHelper.toBoolean(fieldsTb.get("denied"));
      msg.deletedDest =
        ConversionHelper.toBoolean(fieldsTb.get("deletedDest"));
      msg.expired = ConversionHelper.toBoolean(fieldsTb.get("expired"));
      msg.notWriteable =
        ConversionHelper.toBoolean(fieldsTb.get("notWriteable"));
      msg.undeliverable =
        ConversionHelper.toBoolean(fieldsTb.get("undeliverable"));

      msg.optionalHeader = (Hashtable) h.get("optionalHeader");
      msg.properties = (Hashtable) h.get("properties");
      msg.pin = true;
    }
    // Should never happen!
    catch (MessageValueException exc) {}
  
    return msg;
  }

  public void save(String id) {
    if (MessageTracing.dbgMessage.isLoggable(BasicLevel.DEBUG))
      MessageTracing.dbgMessage.log(BasicLevel.DEBUG,
                                    "Message.save(" + id + ')');

    if (! getPersistent()) 
      return;

    if (messagePersistent == null)
      messagePersistent = new MessagePersistent(this);
    messagePersistent.setPin(true);
    setSaveName(MessagePersistenceModule.getSaveName(id,messagePersistent));
    MessagePersistenceModule.save(id,messagePersistent);

    if (MessageTracing.dbgMessage.isLoggable(BasicLevel.DEBUG))
      MessageTracing.dbgMessage.log(BasicLevel.DEBUG,
                                    "Message.save : saveName=" + saveName);
  }

  public void delete() {
    if (MessageTracing.dbgMessage.isLoggable(BasicLevel.DEBUG))
      MessageTracing.dbgMessage.log(BasicLevel.DEBUG,
                                    "Message.delete()");

    if (! getPersistent()) 
      return;
    if (messagePersistent == null)
      messagePersistent = new MessagePersistent(this);
    MessagePersistenceModule.delete(messagePersistent);
  }
  
  private static void writeString(ObjectOutput os,
                                  String s) throws IOException {
    if (s == null) {
      os.writeInt(-1);
    } else if (s.length() == 0) {
      os.writeInt(0);
    } else {
      byte[] bout = s.getBytes();
      os.writeInt(bout.length);
      os.write(bout);
    }
  }

  private static String readString(ObjectInput is) throws IOException {
    int length = is.readInt();
    if (length == -1) {
      return null;
    } else if (length == 0) {
      return "";
    } else {
      byte[] bin = new byte[length];
      is.readFully(bin);
      return new String(bin);
    }
  }

  final static byte BOOLEAN = 0;
  final static byte BYTE = 1;
  final static byte DOUBLE = 2;
  final static byte FLOAT = 3;
  final static byte INTEGER = 4;
  final static byte LONG = 5;
  final static byte SHORT = 6;
  final static byte STRING = 7;

  private static void writeProperties(ObjectOutput os,
                                      Hashtable h) throws IOException {
    if (h == null) {
      os.writeInt(0);
      return;
    }

    os.writeInt(h.size());
    for (Enumeration e = h.keys() ; e.hasMoreElements() ;) {
      String key = (String) e.nextElement();
      writeString(os, key);
      Object value = h.get(key);
      if (value instanceof Boolean) {
        os.writeByte(BOOLEAN);
        os.writeBoolean(((Boolean) value).booleanValue());
      } else if (value instanceof Byte) {
        os.writeByte(BYTE);
        os.writeByte(((Byte) value).byteValue());
      } else if (value instanceof Double) {
        os.writeByte(DOUBLE);
        os.writeDouble(((Double) value).doubleValue());
      } else if (value instanceof Float) {
        os.writeByte(FLOAT);
        os.writeFloat(((Float) value).floatValue());
      } else if (value instanceof Integer) {
        os.writeByte(INTEGER);
        os.writeInt(((Integer) value).intValue());
      } else if (value instanceof Long) {
        os.writeByte(LONG);
        os.writeLong(((Long) value).longValue());
       } else if (value instanceof Short) {
        os.writeByte(SHORT);
        os.writeShort(((Short) value).shortValue());
      } else if (value instanceof String) {
        os.writeByte(STRING);
        writeString(os, (String) value);
      }
     }
  }

  private static Hashtable readProperties(ObjectInput is) throws IOException, ClassNotFoundException {
    int size = is.readInt();
    if (size == 0) return null;

    Hashtable h = new Hashtable(size);
    
    for (int i = 0; i < size; i++) {
      String key = readString(is);
      byte type = is.readByte();
      if (type == BOOLEAN) {
        h.put(key, new Boolean(is.readBoolean()));
      } else if (type == BYTE) {
        h.put(key, new Byte(is.readByte()));
      } else if (type == DOUBLE) {
        h.put(key, new Double(is.readDouble()));
      } else if (type == FLOAT) {
        h.put(key, new Float(is.readFloat()));
      } else if (type == INTEGER) {
        h.put(key, new Integer(is.readInt()));
      } else if (type == LONG) {
        h.put(key, new Long(is.readLong()));
      } else if (type == SHORT) {
        h.put(key, new Short(is.readShort()));
      } else if (type == STRING) {
        h.put(key, readString(is));
      }
    }
    return h;
  }

  /**
   * Specializes the serialization method for protecting the message's
   * properties and body as soon as it is sent.
   */
  private void writeObject(ObjectOutputStream os) 
//   public void writeExternal(ObjectOutput os)
    throws IOException {
    os.writeInt(type);
    os.writeBoolean(persistent);
    os.writeBoolean(noBody);
    writeString(os, id);
    os.writeInt(priority);
    os.writeLong(expiration);
    os.writeLong(timestamp);
    writeString(os, toId);
    writeString(os, toType);
    writeString(os, correlationId);
    writeString(os, replyToId);
    writeString(os, replyToType);

    os.writeObject(optionalHeader);

    os.writeObject(getMessageBody());
    os.writeBoolean(bodyRO);

    writeProperties(os, properties);

    os.writeLong(order);
    os.writeInt(deliveryCount);

    os.writeBoolean(denied);
    os.writeBoolean(deletedDest);
    os.writeBoolean(expired);
    os.writeBoolean(notWriteable);
    os.writeBoolean(undeliverable);

    bodyRO = true;
  }

  /**
   * Specializes the deserialization method for initializing the message's
   * transient fields.
   */
  private void readObject(ObjectInputStream is)
//   public void readExternal(ObjectInput is)
    throws IOException, ClassNotFoundException {
    type = is.readInt();
    persistent = is.readBoolean();
    noBody = is.readBoolean();
    id = readString(is);
    priority = is.readInt();
    expiration = is.readLong();
    timestamp = is.readLong();
    toId = readString(is);
    toType = readString(is);
    correlationId = readString(is);
    replyToId = readString(is);
    replyToType = readString(is);

    optionalHeader = (Hashtable) is.readObject();

    body = (MessageBody) is.readObject();
    bodyRO = is.readBoolean();

    properties = readProperties(is);
    
    order = is.readLong();
    deliveryCount = is.readInt();

    denied = is.readBoolean();
    deletedDest= is.readBoolean();
    expired = is.readBoolean();
    notWriteable = is.readBoolean();
    undeliverable = is.readBoolean();

    acksCounter = 0;
    durableAcksCounter = 0;
    propertiesRO = true;
    pin = true;
  }

  public String toString() {
    StringBuffer buff = new StringBuffer();
    buff.append('(');
    buff.append(super.toString());
    buff.append(",type=");
    buff.append(type);
    buff.append(",id=");
    buff.append(id);
    buff.append(",toId=");
    buff.append(toId);
    buff.append(",toType=");
    buff.append(toType);
    buff.append(",replyToId=");
    buff.append(replyToId);
    buff.append(",replyToType=");
    buff.append(replyToType);
    buff.append(",pin=");
    buff.append(pin);
    buff.append(",body=");
    buff.append(getMessageBody());
    buff.append(')');
    return buff.toString();
  }
}