jmssession.java

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

    }

    /**
     * Receive the next message that arrives within the specified timeout
     * interval. This call blocks until a message arrives, the timeout expires,
     * or this message consumer is closed. A timeout of <code>0</code> never
     * expires and the call blocks indefinitely.
     *
     * @param consumerId the consumer identifier
     * @param timeout    the timeout interval, in milliseconds
     * @return the next message produced for the consumer, or <code>null</code>
     *         if the timeout expires or the consumer concurrently closed
     * @throws JMSException if the next message can't be received
     */
    public MessageImpl receive(long consumerId, long timeout)
            throws JMSException {
        MessageImpl message = null;
        ensureOpen();

        synchronized (_receiveLock) {
            if (_blockingConsumer != -1) {
                throw new IllegalStateException(
                        "Session cannot be accessed concurrently");
            }

            _blockingConsumer = consumerId;

            long start = (timeout != 0) ? System.currentTimeMillis() : 0;
            try {
                while (message == null && !isClosed()) {
                    if (timeout == 0) {
                        message = getServerSession().receive(consumerId, 0);
                    } else {
                        message = getServerSession().receive(consumerId,
                                                             timeout);
                    }
                    if (message == null && !isClosed()) {
                        // no message received in the required time.
                        // Wait for a notification from the server that
                        // a message has become available.
                        try {
                            if (timeout == 0) {
                                _receiveLock.wait();
                            } else {
                                long elapsed = System.currentTimeMillis()
                                        - start;
                                if (elapsed >= timeout) {
                                    // no message received in the required time
                                    break;
                                } else {
                                    // adjust the timeout so that the client
                                    // only waits as long as the original
                                    // timeout
                                    timeout -= elapsed;
                                }
                                _receiveLock.wait(timeout);
                            }
                        } catch (InterruptedException ignore) {
                            // no-op
                        }
                    }
                }

                if (message != null) {
                    message.setSession(this);
                    if (_ackMode == AUTO_ACKNOWLEDGE
                            || _ackMode == DUPS_OK_ACKNOWLEDGE) {
                        getServerSession().acknowledgeMessage(
                                message.getConsumerId(),
                                message.getMessageId().toString());
                    }
                }
            } finally {
                _blockingConsumer = -1;
            }
        }
        return message;
    }

    /**
     * Receive the next message if one is immediately available.
     *
     * @param consumerId the consumer identifier
     * @return the next message produced for this consumer, or <code>null</code>
     *         if one is not available
     * @throws JMSException if the next message can't be received
     */
    public MessageImpl receiveNoWait(long consumerId) throws JMSException {
        ensureOpen();
        MessageImpl message = getServerSession().receiveNoWait(consumerId);
        if (message != null) {
            message.setSession(this);
            if (_ackMode == AUTO_ACKNOWLEDGE
                    || _ackMode == DUPS_OK_ACKNOWLEDGE) {
                getServerSession().acknowledgeMessage(
                        message.getConsumerId(),
                        message.getMessageId().toString());
            }
        }
        return message;
    }

    /**
     * Browse up to count messages.
     *
     * @param consumerId the consumer identifier
     * @param count      the maximum number of messages to receive
     * @return a list of {@link MessageImpl} instances
     * @throws JMSException for any JMS error
     */
    public List browse(long consumerId, int count)
            throws JMSException {
        ensureOpen();
        return getServerSession().browse(consumerId, count);
    }

    /**
     * Send the specified message to the server.
     *
     * @param message the message to send
     * @throws JMSException if the message can't be sent
     */
    protected void sendMessage(Message message) throws JMSException {
        if (getTransacted()) {
            // if the session is transacted then cache the message locally.
            // and wait for a commit or a rollback
            if (message instanceof MessageImpl) {
                try {
                    message = (Message) ((MessageImpl) message).clone();
                } catch (CloneNotSupportedException error) {
                    throw new JMSException(error.getMessage());
                }
            } else {
                message = convert(message);
            }
            _messagesToSend.add(message);
        } else {
            if (!(message instanceof MessageImpl)) {
                message = convert(message);
            }
            getServerSession().send((MessageImpl) message);
        }
    }

    /**
     * Returns the server session.
     *
     * @return the server session
     */
    protected ServerSession getServerSession() {
        return _session;
    }

    /**
     * Return a reference to the connection that created this session.
     *
     * @return the owning connection
     */
    protected JmsConnection getConnection() {
        return _connection;
    }

    /**
     * Creates a new message consumer, returning its identity.
     *
     * @param destination the destination to access
     * @param selector    the message selector. May be <code>null</code>
     * @param noLocal     if true, and the destination is a topic, inhibits the
     *                    delivery of messages published by its own connection.
     *                    The behavior for <code>noLocal</code> is not specified
     *                    if the destination is a queue.
     * @throws JMSException                if the session fails to create a
     *                                     MessageConsumer due to some internal
     *                                     error.
     * @throws InvalidDestinationException if an invalid destination is
     *                                     specified.
     * @throws InvalidSelectorException    if the message selector is invalid.
     */
    protected long allocateConsumer(Destination destination,
                                    String selector, boolean noLocal)
            throws JMSException {
        ensureOpen();

        if (!(destination instanceof JmsDestination)) {
            throw new InvalidDestinationException("Cannot create MessageConsumer for destination="
                                                  + destination);
        }
        JmsDestination dest = (JmsDestination) destination;

        // check to see if the destination is temporary. A temporary destination
        // can only be used within the context of the owning connection
        if (!checkForValidTemporaryDestination(dest)) {
            throw new InvalidDestinationException(
                    "Trying to create a MessageConsumer for a temporary "
                    + "destination that is not bound to this connection");
        }

        return _session.createConsumer(dest, selector, noLocal);
    }

    /**
     * This method checks the destination. If the destination is not temporary
     * then return true. If it is a temporary destination and it is owned by
     * this session's connection then it returns true. If it is a tmeporary
     * destination and it is owned by another connection then it returns false
     *
     * @param destination the destination to check
     * @return <code>true</code> if the destination is valid
     */
    protected boolean checkForValidTemporaryDestination(
            JmsDestination destination) {
        boolean result = false;

        if (destination.isTemporaryDestination()) {
            JmsTemporaryDestination temp =
                    (JmsTemporaryDestination) destination;

            // check  that this temp destination is owned by the session's
            // connection.
            if (temp.validForConnection(getConnection())) {
                result = true;
            }
        } else {
            result = true;
        }

        return result;
    }

    /**
     * Add a consumer to the list of consumers managed by this session.
     *
     * @param consumer the consumer to add
     */
    protected void addConsumer(JmsMessageConsumer consumer) {
        _consumers.put(new Long(consumer.getConsumerId()), consumer);
    }

    /**
     * Remove a consumer, deregistering it on the server.
     *
     * @param consumer the consumer to remove
     * @throws JMSException if removal fails
     */
    protected void removeConsumer(JmsMessageConsumer consumer)
            throws JMSException {
        long consumerId = consumer.getConsumerId();
        try {
            _session.closeConsumer(consumerId);
        } finally {
            _consumers.remove(new Long(consumerId));
        }
    }

    /**
     * Add a producer to the list of producers managed by this session.
     *
     * @param producer the producer to add
     */
    protected void addProducer(JmsMessageProducer producer) {
        _producers.add(producer);
    }

    /**
     * Remove the producer from the list of managed producers.
     *
     * @param producer the producer to remove
     */
    protected void removeProducer(JmsMessageProducer producer) {
        _producers.remove(producer);
    }

    /**
     * Check if the session is closed.
     *
     * @return <code>true</code> if the session is closed
     */
    protected final boolean isClosed() {
        return _closed;
    }

    /**
     * Add a message to the message cache. This message will be processed when
     * the run() method is called.
     *
     * @param message the message to add.
     */
    protected void addMessage(Message message) {
        _messageCache.add(message);
    }

    /**
     * Verifies that the session isn't closed.
     *
     * @throws IllegalStateException if the session is closed
     */
    protected void ensureOpen() throws IllegalStateException {
        if (isClosed()) {
            throw new IllegalStateException(
                    "Cannot perform operation - session has been closed");
        }
    }

    /**
     * Verifies that the session is transactional.
     *
     * @throws IllegalStateException if the session isn't transactional
     */
    private void ensureTransactional() throws IllegalStateException {
        if (_ackMode != SESSION_TRANSACTED) {
            throw new IllegalStateException(
                    "Cannot perform operatiorn - session is not transactional");
        }
    }

    /**
     * Notifies any blocking synchronous consumer.
     */
    private void notifyConsumer() {
        synchronized (_receiveLock) {
            _receiveLock.notifyAll();
        }
    }

    /**
     * Convert a message to its corresponding OpenJMS implementation.
     *
     * @param message the message to convert
     * @return the OpenJMS implementation of the message
     * @throws JMSException for any error
     */
    private Message convert(Message message) throws JMSException {
        MessageConverter converter =
                MessageConverterFactory.create(message);
        return converter.convert(message);
    }

}