jmssession.java

来自「OpenJMS是一个开源的Java Message Service API 1.」· Java 代码 · 共 1,363 行 · 第 1/4 页

        JmsQueueBrowser browser = new JmsQueueBrowser(this, consumerId, queue,
                                                      messageSelector);
        addConsumer(browser);
        return browser;
    }

    /**
     * Creates a <code>TemporaryQueue</code> object. Its lifetime will be that
     * of the <code>Connection</code> unless it is deleted earlier.
     *
     * @return a temporary queue identity
     * @throws JMSException if the session fails to create a temporary queue due
     *                      to some internal error.
     */
    public TemporaryQueue createTemporaryQueue() throws JMSException {
        ensureOpen();
        return JmsTemporaryQueue.create(getConnection());
    }

    /**
     * Creates a <code>TemporaryTopic</code> object. Its lifetime will be that
     * of the <code>Connection</code> unless it is deleted earlier.
     *
     * @return a temporary topic identity
     * @throws JMSException if the session fails to create a temporary topic due
     *                      to some internal error.
     */
    public TemporaryTopic createTemporaryTopic() throws JMSException {
        ensureOpen();
        return JmsTemporaryTopic.create(getConnection());
    }

    /**
     * Unsubscribes a durable subscription that has been created by a client.
     * <p/>
     * <P>This method deletes the state being maintained on behalf of the
     * subscriber by its provider.
     * <p/>
     * <P>It is erroneous for a client to delete a durable subscription while
     * there is an active <code>MessageConsumer</code> or
     * <code>TopicSubscriber</code> for the subscription, or while a consumed
     * message is part of a pending transaction or has not been acknowledged in
     * the session.
     *
     * @param name the name used to identify this subscription
     * @throws JMSException                if the session fails to unsubscribe
     *                                     to the durable subscription due to
     *                                     some internal error.
     * @throws InvalidDestinationException if an invalid subscription name is
     *                                     specified.
     */
    public void unsubscribe(String name) throws JMSException {
        ensureOpen();
        _session.unsubscribe(name);
    }

    /**
     * Commit all messages done in this transaction
     *
     * @throws JMSException if the transaction cannot be committed
     */
    public void commit() throws JMSException {
        ensureOpen();
        ensureTransactional();

        // send all the cached messages to the server
        getServerSession().send(_messagesToSend);
        _messagesToSend.clear();

        // commit the session
        getServerSession().commit();
    }

    /**
     * Rollback any messages done in this transaction
     *
     * @throws JMSException if the transaction cannot be rolled back
     */
    public void rollback() throws JMSException {
        ensureOpen();
        ensureTransactional();

        // clear all the cached messages
        _messagesToSend.clear();

        // rollback the session
        getServerSession().rollback();
    }

    /**
     * Close the session. This call will block until a receive or message
     * listener in progress has completed. A blocked message consumer receive
     * call returns <code>null</code> when this session is closed.
     *
     * @throws JMSException if the session can't be closed
     */
    public void close() throws JMSException {
        boolean closing;
        synchronized (_closeLock) {
            closing = _closing;
            _closing = true;
        }
        if (!closing) {
            // must stop first to ensure any active listener completes
            stop();

            // wake up any blocking consumer
            synchronized (_receiveLock) {
                _receiveLock.notifyAll();
            }

            _closed = true;

            // close the producers
            JmsMessageProducer[] producers =
                    (JmsMessageProducer[]) _producers.toArray(
                            new JmsMessageProducer[0]);
            for (int i = 0; i < producers.length; ++i) {
                JmsMessageProducer producer = producers[i];
                producer.close();
            }

            // close the consumers
            JmsMessageConsumer[] consumers =
                    (JmsMessageConsumer[]) _consumers.values().toArray(
                            new JmsMessageConsumer[0]);
            for (int i = 0; i < consumers.length; ++i) {
                JmsMessageConsumer consumer = consumers[i];
                consumer.close();
            }

            // deregister this with the connection
            _connection.removeSession(this);
            _connection = null;

            // clear any cached messages
            _messagesToSend.clear();

            // issue a close to the remote session. This will release any
            // allocated remote resources
            getServerSession().close();
            _session = null;
        }
    }

    /**
     * Stop message delivery in this session, and restart sending messages with
     * the oldest unacknowledged message
     *
     * @throws JMSException if the session can't be recovered
     */
    public void recover() throws JMSException {
        ensureOpen();
        if (getTransacted()) {
            throw new IllegalStateException(
                    "Cannot recover from a transacted session");
        }

        getServerSession().recover();
    }

    /**
     * Returns the message listener associated with the session
     *
     * @return the message listener associated with the session, or
     *         <code>null</code> if no listener is registered
     * @throws JMSException if the session is closed
     */
    public MessageListener getMessageListener() throws JMSException {
        ensureOpen();
        return _listener;
    }

    /**
     * Sets the session's message listener.
     *
     * @param listener the session's message listener
     * @throws JMSException if the session is closed
     */
    public void setMessageListener(MessageListener listener)
            throws JMSException {
        ensureOpen();
        _listener = listener;
    }

    /**
     * Iterates through the list of messages added by an {@link
     * JmsConnectionConsumer}, sending them to the registered listener
     */
    public void run() {
        try {
            while (!_messageCache.isEmpty()) {
                Message message = (Message) _messageCache.remove(0);
                _listener.onMessage(message);
            }
        } catch (Exception exception) {
            _log.error("Error in the Session.run()", exception);
        } finally {
            // Clear message cache
            _messageCache.clear();
        }
    }

    /**
     * Set the message listener for a particular consumer.
     * <p/>
     * If a listener is already registered for the consumer, it will be
     * automatically overwritten
     *
     * @param listener the message listener
     * @throws JMSException if the listener can't be set
     */
    public void setMessageListener(JmsMessageConsumer listener)
            throws JMSException {
        ensureOpen();
        setAsynchronous(listener.getConsumerId(), true);
    }

    /**
     * Remove a message listener
     *
     * @param listener the message listener to remove
     * @throws JMSException if the listener can't be removed
     */
    public void removeMessageListener(JmsMessageConsumer listener)
            throws JMSException {
        ensureOpen();
        setAsynchronous(listener.getConsumerId(), false);
    }

    /**
     * This will start message delivery to this session. If message delivery has
     * already started, or the session is currently being closed then this is a
     * no-op.
     *
     * @throws JMSException if message delivery can't be started
     */
    public void start() throws JMSException {
        ensureOpen();
        synchronized (_closeLock) {
            if (_stopped && !_closing) {
                getServerSession().start();
                _stopped = false;
            }
        }
    }

    /**
     * This will stop message delivery to this session. If message delivery has
     * already stoped then this is a no-op.
     *
     * @throws JMSException if message delivery can't be stopped
     */
    public void stop() throws JMSException {
        ensureOpen();
        synchronized (_closeLock) {
            if (!_stopped) {
                getServerSession().stop();
                _stopped = true;
            }
        }
    }

    /**
     * Acknowledge the specified message. This is only applicable for
     * CLIENT_ACKNOWLEDGE sessions. For other session types, the request is
     * ignored.
     * <p/>
     * Acking a message automatically acks all those that have come before it.
     *
     * @param message the message to acknowledge
     * @throws JMSException if the message can't be acknowledged
     */
    public void acknowledgeMessage(Message message) throws JMSException {
        ensureOpen();
        if (_ackMode == Session.CLIENT_ACKNOWLEDGE) {
            MessageImpl impl = (MessageImpl) message;
            getServerSession().acknowledgeMessage(impl.getConsumerId(),
                                                  impl.getAckMessageID());
        }
    }

    /**
     * Enable or disable asynchronous message delivery for the specified
     * consumer.
     *
     * @param consumerId the consumer identifier
     * @param enable     <code>true</code> to enable; <code>false</code> to
     *                   disable
     * @throws JMSException if message delivery cannot be enabled or disabled
     */
    public void setAsynchronous(long consumerId, boolean enable)
            throws JMSException {
        ensureOpen();
        getServerSession().setAsynchronous(consumerId, enable);
    }

    /**
     * Deliver a message.
     *
     * @param message the message to deliver
     * @return <code>true</code> if the message was delivered; otherwise
     *         <code>false</code>.
     */
    public boolean onMessage(MessageImpl message) {
        boolean delivered = false;
        message.setJMSXRcvTimestamp(System.currentTimeMillis());

        long consumerId = message.getConsumerId();
        JmsMessageConsumer consumer
                = (JmsMessageConsumer) _consumers.get(new Long(consumerId));
        // tag the session that received this message
        message.setSession(this);
        if (consumer != null) {
            // if a listener is defined for the session then send all the
            // messages to that listener regardless if any consumers are
            // have registered listeners...bit confusing but this is what
            // I believe it should do
            if (_listener != null) {
                try {
                    _listener.onMessage(message);
                    delivered = true;
                } catch (Throwable exception) {
                    _log.error("MessageListener threw exception", exception);
                }
            } else {
                delivered = consumer.onMessage(message);
            }
        } else {
            _log.error("Received a message for an inactive consumer");
        }
        return delivered;
    }

    /**
     * Inform the session that there is a message available for a synchronous
     * consumer.
     */
    public void onMessageAvailable() {
        // wake up any blocking consumer
        notifyConsumer();