persistentmessagehandle.java

实现了Jms的服务器源码,支持多种适配器,DB,FTP,支持多种数据库

     * Set the message expiry time
     *
     * @long time - time in ms since epoch
     */
    public void setExpiryTime(long time) {
        _expiryTime = time;
    }

    /**
     * Return the message expiry time
     *
     * @return long - time in ms since epoch
     */
    public long getExpiryTime() {
        return _expiryTime;
    }

    /**
     * Determines if the message has expired
     *
     * @return <code>true</code> if the message has expired, otherwise
     * <code>false</code>
     */
    public boolean hasExpired() {
        return (_expiryTime != 0 && _expiryTime <= System.currentTimeMillis());
    }

    // implementation of MessageHandle.setConsumerName
    public void setConsumerName(String name) {
        _consumerName = name;
    }

    // implementation of MessageHandle.getConsumerName
    public String getConsumerName() {
        return _consumerName;
    }

    // implementation of MessageHandle.isPersistent
    public boolean isPersistent() {
        return true;
    }

    /**
     * Set the destination that owns this handle
     *
     * @param destination - the destination
     */
    public void setDestination(JmsDestination destination) {
        _destination = destination;
    }

    /**
     * Return the destination for this handle
     *
     * @return JmsDestination
     */
    public JmsDestination getDestination() {
        return _destination;
    }

    /**
     * Set the client id, that owns this handle
     *
     * @param clientId - client identity
     */
    public void setClientId(long clientId) {
        _clientId = clientId;
    }

    /**
     * Retrieve the client identity associated with this handle
     *
     * @return long
     */
    public long getClientId() {
        return _clientId;
    }

    /**
     * Override the behaviour of the base class. I don't think we want to
     * clear the handle unless there are definitely no more references.
     */
    public void clear() {
    }

    /**
     * Destroy this handle
     */
    public void destroy() {
        // notify the message manager
        MessageMgr.instance().handleDestroyed(this);

        // remove the handle from the database
        Connection connection = null;
        TransactionManager tm = null;

        try {
            connection = DatabaseService.getConnection();
            DatabaseService.getAdapter().removeMessageHandle(connection, this);
            connection.commit();
        } catch (PersistenceException exception) {
            if (connection != null) {
                try {
                    connection.rollback();
                } catch (Exception nested) {
                    // ignore
                }
            }
        } catch (Exception exception) {
            // ignore for the moment
        } finally {
            if (connection != null) {
                try {
                    connection.close();
                } catch (Exception nested) {
                    // ignore
                }
            }
        }
    }

    /**
     * Clone the persistent message handle object
     *
     * @return copy of a PersistentMessageHandle
     * @throws CloneNotSupportedException
     */
    public Object clone() throws CloneNotSupportedException {
        PersistentMessageHandle handle =
            (PersistentMessageHandle) super.clone();

        handle._id = new MessageId(_id.getId());
        handle._priority = _priority;
        handle._delivered = _delivered;
        handle._acceptedTime = _acceptedTime;
        handle._expiryTime = _expiryTime;
        handle._consumerName = _consumerName;
        handle._sequenceNumber = _sequenceNumber;

        // clone the destination
        if (_destination instanceof JmsQueue) {
            handle._destination = new JmsQueue(_destination.getName());
        } else {
            handle._destination = new JmsTopic(_destination.getName());
        }
        handle._destination.setPersistent(true);

        return handle;
    }

    // override the definition of equals
    public boolean equals(Object object) {

        boolean result = false;

        if ((object != null) &&
            (object instanceof PersistentMessageHandle)) {

            PersistentMessageHandle mhdl = (PersistentMessageHandle) object;
            if (mhdl._id.equals(_id)) {
                result = true;
            }

            //if ((mhdl._id.equals(_id)) &&
            //    (mhdl._consumerName.equals(_consumerName))) {
            //    result = true;
            //}
        }

        return result;
    }

    // override the Object.toString()
    public String toString() {
        return ("Handle : " + _id.getId() + "@" + _consumerName + ":" +
            _sequenceNumber + ":" + _acceptedTime + ":" + _priority);
    }


    // override the definition of hashCode
    public int hashCode() {
        if (_hashCode == -1) {
            _hashCode = (_id + _consumerName).hashCode();
        }

        return _hashCode;
    }

    /**
     * This is used to set the message as delivered. This can only be called
     * by the delivery engine and should not be used anywhere else. This will
     * mark the message as being delivered and also update the database
     * <p>
     * We do not really throw an exception here....and it seems that this is
     * a very fine grained approach to dealing with this...basically translates to
     * a transaction for every message.
     */
    public void setDelivered() {
        if (!_delivered) {

            Connection connection = null;
            TransactionManager tm = null;

            try {
                connection = DatabaseService.getConnection();

                DatabaseService.getAdapter().updateMessageHandle(connection,
                    this);
                _delivered = true;

                // commit the transactions
                connection.commit();
            } catch (PersistenceException exception) {
                System.err.println("Failed in setDelivereds b/c " +
                    exception.toString());
                if (connection != null) {
                    try {
                        connection.rollback();
                    } catch (Exception nested) {
                        // ignore
                    }
                }
            } catch (Exception exception) {
                // ignore for the moment
            } finally {
                if (connection != null) {
                    try {
                        connection.close();
                    } catch (Exception nested) {
                        // ignore
                    }
                }
            }
        }
    }

    // implementation of Externalizable.writeExternal
    public void writeExternal(ObjectOutput stream)
        throws IOException {
        stream.writeLong(serialVersionUID);
        stream.writeObject(_id);
        stream.writeInt(_priority);
        stream.writeBoolean(_delivered);
        stream.writeLong(_acceptedTime);
        stream.writeLong(_sequenceNumber);
        stream.writeLong(_expiryTime);
        stream.writeObject(_consumerName);
        stream.writeObject(_destination);
        super.writeExternal(stream);
    }

    // implementation of Externalizable.writeExternal
    public void readExternal(ObjectInput stream)
        throws IOException, ClassNotFoundException {
        long version = stream.readLong();
        if (version == serialVersionUID) {
            _id = (MessageId) stream.readObject();
            _priority = stream.readInt();
            _delivered = stream.readBoolean();
            _acceptedTime = stream.readLong();
            _sequenceNumber = stream.readLong();
            _expiryTime = stream.readLong();
            _consumerName = (String) stream.readObject();
            _destination = (JmsDestination) stream.readObject();
            super.readExternal(stream);
        } else {
            throw new IOException("PersistentMessageHandle with version " +
                version + " is not supported.");
        }
    }
}