messagevector.java

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

        removeMessageAt(i);
        validated -= 1;
        return;
      }
    }

    logmon.log(BasicLevel.ERROR,
               logmsg + "removeMessage #" + msg.getStamp() + " not found");

    return;
  }

  /**
   *  Removes all messages with a stamp less than the specified one.
   * Be careful with the use of this method, in particular it does not
   * take in account the multiples incoming nodes.
   */
  synchronized int remove(int stamp) {
    if (validated == 0) return 0;
    
    if (Debug.debug && logmon.isLoggable(BasicLevel.DEBUG))
      logmon.log(BasicLevel.DEBUG, logmsg + "remove #" + stamp);

    int i = 0;
    for (; i<validated; i++) {
      Message msg = getMessageAt(i);
      if (stamp < msg.getStamp()) break;
    }

    for (int j=0; j<i; j++) {
      removeMessageAt(0);
    }
    validated -= i;
    
    if (Debug.debug && logmon.isLoggable(BasicLevel.DEBUG))
      logmon.log(BasicLevel.DEBUG, logmsg + "remove #" + stamp + " ->" +i);

    return i;
  }

  /**
   * Inserts the specified message to this <code>MessageVector</code> at
   * the specified index. Each component in this vector with an index greater
   * or equal to the specified index is shifted upward.
   *
   * @param item	the message to be pushed onto this queue.
   * @param index	where to insert the new message.
   */
  void insertMessageAt(Message item, int index) {
    if (Debug.debug && logmon.isLoggable(BasicLevel.DEBUG))
      logmon.log(BasicLevel.DEBUG,
                 logmsg + "insertMessageAt(" + item + ", " + index + ")");

    if (count == data.length) {
      Object newData[] = new Object[data.length *2];
      if ((first + count) < data.length) {
        System.arraycopy(data, first, newData, 0, count);
      } else {
        int j = data.length - first;
        System.arraycopy(data, first, newData, 0, j);
        System.arraycopy(data, 0, newData, j, count - j);
      }
      first = 0;
      data = newData;
    }
    if (index != count)
      System.arraycopy(data, index, data, index + 1, count - index);
    if (persistent)
      data[(first + index)%data.length] = new MessageSoftRef(item);
    else
      data[(first + index)%data.length] = item;
    count += 1;

    if (Debug.debug && logmon.isLoggable(BasicLevel.DEBUG))
      logmon.log(BasicLevel.DEBUG,
                 logmsg + "insertMessageAt() -> " + this);
  }

  /**
   * Adds the specified message to the end of internal <code>Vector</code>.
   *
   * @param   item   the message to be added onto this queue.
   */
  void addMessage(Message item) {
    if (Debug.debug && logmon.isLoggable(BasicLevel.DEBUG))
      logmon.log(BasicLevel.DEBUG,
                 logmsg + "addMessage(" + item + ")");

    insertMessageAt(item, count);
  }

  /**
   * Returns the message at the specified index.
   *
   * @param index	the index of the message.
   * @return     	The message at the top of this queue.
   */
  Message getMessageAt(int index) {
    if (Debug.debug && logmon.isLoggable(BasicLevel.DEBUG))
      logmon.log(BasicLevel.DEBUG, logmsg + "getMessageAt(" + index + ")");

    int idx = (first + index)%data.length;
    if (persistent) {
      Message msg = ((MessageSoftRef) data[idx]).getMessage();
      if (msg == null) {
        msg = ((MessageSoftRef) data[idx]).loadMessage();
        data[idx] = new MessageSoftRef(msg);
      }
      return msg;
    } else {
      return (Message) data[idx];
    }
  }

  /**
   * Deletes the message at the specified index.
   *
   * @param index	the index of the message to remove.
   */
  void removeMessageAt(int index) {
    if ((first + index) < data.length) {
      // Moves the start of the vector +1 to the empty 'box'
      System.arraycopy(data, first,
                       data, first +1, index);
      // Erase the old first 'box'
      data[first] = null; /* to let gc do its work */
      // Move the first ptr +1, and decrease counter
      first = (first +1)%data.length;
      count -= 1;
    } else {
      // Moves the end of the vector -1 to the empty 'box'
      System.arraycopy(data, (first + index)%data.length +1,
                       data, (first + index)%data.length, count - index -1);
      // Erase the old last 'box'
      data[(first + count -1)%data.length] = null; /* to let gc do its work */
      // Decrease counter
      count -= 1;
    }
    if (count == 0) first = 0;

    if (Debug.debug && logmon.isLoggable(BasicLevel.DEBUG))
      logmon.log(BasicLevel.DEBUG,
                 logmsg + "removeMessageAt(" + index + ") -> " + this);
  }

  /**
   * Returns the number of messages in this vector.
   *
   * @return  the number of messages in this vector.
   */
  public int size() {
    return count;
  }

  /**
   * Returns a string representation of this <code>MessageVector</code>
   * object. Be careful we scan the vector without synchronization, so the
   * result can be incoherent.
   *
   * @return	A string representation of this object. 
   */
  public String toString() {
    StringBuffer strbuf = new StringBuffer();
    
    strbuf.append('(').append(super.toString());
    strbuf.append(",first=").append(first);
    strbuf.append(",count=").append(count);
    strbuf.append(",validated=").append(validated).append(",(");
    for (int i=0; i<data.length; i++) {
      strbuf.append(data[i]).append(',');
    }
    strbuf.append("))");
    
    return strbuf.toString();
  }
  
  final class MessageSoftRef extends java.lang.ref.SoftReference {
    /**
     *  Name for persistent message, used to retrieve garbaged message
     * from persistent storage.
     */
    String name = null;
    /**
     *  Reference for transient message, used to pin non persistent
     * in memory.
     */
    Message ref = null;
    
    MessageSoftRef(Message msg) {
      super(msg);
      if (msg.isPersistent())
        name = msg.toStringId();
      else
        ref = msg;
    }

    /**
     * Returns this reference message's referent. If the message has been
     * swap out it returns null.
     *
     * @return The message to which this reference refers.
     */
    public Message getMessage() {
      if (ref != null) return ref;
      return (Message) get();
    }

    /**
     * Loads from disk this reference message's referent if the message
     * has been swap out. It should be called only after a getMessage
     * returning null.
     *
     * @return The message to which this reference refers.
     */
    public Message loadMessage() throws TransactionError {
      if (ref != null) return ref;

      Message msg;
      try {
        msg = Message.load(name);

        if (logmon.isLoggable(BasicLevel.DEBUG))
          logmon.log(BasicLevel.DEBUG, logmsg + "reload from disk " + msg);
      } catch (Exception exc) {
        logmon.log(BasicLevel.ERROR,
                   logmsg + "Can't load message " + name, exc);
        throw new TransactionError(exc);
      }
      return msg;
    }

    /**
     * Returns a string representation of this <code>MessageSoftRef</code>
     * object.
     *
     * @return	A string representation of this object. 
     */
    public String toString() {
      StringBuffer strbuf = new StringBuffer();
      
      strbuf.append('(').append(super.toString());
      strbuf.append(",name=").append(name);
      strbuf.append(",ref=").append(ref);
      strbuf.append("))");
      
      return strbuf.toString();
    }
  }

  final class TransactionError extends Error {
    TransactionError(Throwable cause) {
      super(cause);
    }
  }
}