sessionconsumer.java

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

/**
 * Redistribution and use of this software and associated documentation
 * ("Software"), with or without modification, are permitted provided
 * that the following conditions are met:
 *
 * 1. Redistributions of source code must retain copyright
 *    statements and notices.  Redistributions must also contain a
 *    copy of this document.
 *
 * 2. Redistributions in binary form must reproduce the
 *    above copyright notice, this list of conditions and the
 *    following disclaimer in the documentation and/or other
 *    materials provided with the distribution.
 *
 * 3. The name "Exolab" must not be used to endorse or promote
 *    products derived from this Software without prior written
 *    permission of Exoffice Technologies.  For written permission,
 *    please contact [e-mail].
 *
 * 4. Products derived from this Software may not be called "Exolab"
 *    nor may "Exolab" appear in their names without prior written
 *    permission of Exoffice Technologies. Exolab is a registered
 *    trademark of Exoffice Technologies.
 *
 * 5. Due credit should be given to the Exolab Project
 *    (http://www.exolab.org/).
 *
 * THIS SOFTWARE IS PROVIDED BY EXOFFICE TECHNOLOGIES AND CONTRIBUTORS
 * ``AS IS'' AND ANY EXPRESSED OR IMPLIED WARRANTIES, INCLUDING, BUT
 * NOT LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND
 * FITNESS FOR A PARTICULAR PURPOSE ARE DISCLAIMED.  IN NO EVENT SHALL
 * EXOFFICE TECHNOLOGIES OR ITS CONTRIBUTORS BE LIABLE FOR ANY DIRECT,
 * INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES
 * (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR
 * SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION)
 * HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT,
 * STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
 * ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED
 * OF THE POSSIBILITY OF SUCH DAMAGE.
 *
 * Copyright 2005 (C) Exoffice Technologies Inc. All Rights Reserved.
 *
 * $Id: SessionConsumer.java,v 1.4 2007/01/24 12:00:28 tanderson Exp $
 */
package org.exolab.jms.server;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.exolab.jms.client.JmsMessageListener;
import org.exolab.jms.message.MessageImpl;
import org.exolab.jms.messagemgr.Condition;
import org.exolab.jms.messagemgr.ConsumerEndpoint;
import org.exolab.jms.messagemgr.ConsumerEndpointListener;
import org.exolab.jms.messagemgr.Flag;
import org.exolab.jms.messagemgr.MessageHandle;
import org.exolab.jms.messagemgr.QueueBrowserEndpoint;
import org.exolab.jms.messagemgr.TimedCondition;
import org.exolab.jms.persistence.DatabaseService;
import org.exolab.jms.persistence.PersistenceException;
import org.exolab.jms.scheduler.Scheduler;
import org.exolab.jms.scheduler.SerialTask;

import javax.jms.JMSException;
import java.rmi.RemoteException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.Iterator;
import java.util.LinkedList;
import java.util.List;


/**
 * Manages all consumers for a session.
 *
 * @author <a href="mailto:[e-mail]">Tim Anderson</a>
 * @version $Revision: 1.4 $ $Date: 2007/01/24 12:00:28 $
 */
class SessionConsumer implements ConsumerEndpointListener {

    /**
     * The message listener is the reference to a remote client that will
     * receive the messages.
     */
    private JmsMessageListener _listener;

    /**
     * Maintain a set of ConsumerEndpoint instances, keyed on id.
     */
    private final HashMap _consumers = new HashMap();

    /**
     * Caches all sent messages.
     */
    private final SentMessageCache _sent;

    /**
     * The database service.
     */
    private final DatabaseService _database;

    /**
     * The set of consumer endpoints with messages pending.
     */
    private final LinkedList _pending = new LinkedList();

    /**
     * Determines if the sender is stopping/stopped.
     */
    private Flag _stop = new Flag(true);

    /**
     * Stop/start lock.
     */
    private final Object _restartLock = new Object();

    /**
     * The active consumer lock.
     */
    private final Object _removeLock = new Object();

    /**
     * The consumer currently being dispatched to.
     */
    private long _consumerId = -1;

    /**
     * The maximum number of messages that a dispatch can deliver at any one
     * time
     */
    private final int MAX_MESSAGES = 200;

    /**
     * The logger.
     */
    private static final Log _log = LogFactory.getLog(SessionConsumer.class);


    private final SerialTask _runner;

    /**
     * Construct a new <code>SessionConsumer</code>.
     *
     * @param ackMode   the message acknowledgement mode, or
     *                  <code>Session.TRANSACTED_SESSION</code>
     *                  if the session is transactional
     * @param database  the database service
     * @param scheduler the scheduler
     */
    public SessionConsumer(int ackMode, DatabaseService database,
                           Scheduler scheduler) {
        _database = database;
        _sent = new SentMessageCache(ackMode);
        Runnable task = new Runnable() {
            public void run() {
                dispatch();
            }
        };

        _runner = new SerialTask(task, scheduler);
    }

    /**
     * Set the listener for this session.
     * <p/>
     * The listener is notified whenever a message for the session is present.
     *
     * @param listener the message listener
     */
    public synchronized void setMessageListener(JmsMessageListener listener) {
        _listener = listener;
    }

    /**
     * Register a consumer.
     *
     * @param consumer the consumer to add
     */
    public synchronized void addConsumer(ConsumerEndpoint consumer) {
        final long id = consumer.getId();
        _consumers.put(new Long(id), consumer);
        consumer.setListener(this);
    }

    /**
     * Deregister a consumer.
     *
     * @param consumerId the consumer identifier
     * @return the consumer
     * @throws JMSException if the consumer can't be removed
     */
    public ConsumerEndpoint removeConsumer(long consumerId)
            throws JMSException {
        ConsumerEndpoint consumer;
        synchronized (_removeLock) {
            while (consumerId == _consumerId) {
                try {
                    _removeLock.wait();
                } catch (InterruptedException ignore) {
                    // do nothing
                }
            }
            synchronized (this) {
                consumer = (ConsumerEndpoint) _consumers.remove(
                        new Long(consumerId));
                if (consumer == null) {
                    throw new JMSException("No consumer with id=" + consumerId);
                }
                consumer.setListener(null);
            }
            synchronized (_pending) {
                _pending.remove(consumer);
            }
        }

        return consumer;
    }

    /**
     * Returns the consumers.
     *
     * @return the consumers
     */
    public synchronized ConsumerEndpoint[] getConsumers() {
        return (ConsumerEndpoint[]) _consumers.values()
                .toArray(new ConsumerEndpoint[0]);
    }

    /**
     * Enable or disable asynchronous message delivery for a consumer.
     *
     * @param consumerId the consumer identifier
     * @param enable     <code>true</code> to enable; <code>false</code> to
     *                   disable
     * @throws JMSException for any JMS error
     */
    public void setAsynchronous(long consumerId, boolean enable)
            throws JMSException {
        ConsumerEndpoint consumer = getConsumer(consumerId);
        consumer.setAsynchronous(enable);
        if (enable && consumer.getMessageCount() != 0) {
            messageAvailable(consumer);
        }

    }

    /**
     * Stop message delivery.
     */
    public void stop() {
        synchronized (_restartLock) {
            _stop.set(true);
            _runner.stop();
            _log.debug("stopped delivery");
        }
    }

    /**
     * Start message delivery.
     */
    public void start() throws JMSException {
        synchronized (_restartLock) {
            _log.debug("start");
            _stop.set(false);
            for (Iterator i = _consumers.values().iterator(); i.hasNext();) {
                ConsumerEndpoint consumer = (ConsumerEndpoint) i.next();
                if (needsScheduling(consumer)) {
                    queue(consumer);
                }
            }
            try {
                _runner.schedule();
            } catch (InterruptedException exception) {
                _log.error("Failed to start worker", exception);
                throw new JMSException("Failed to start worker: " + exception);
            }
        }
    }

    /**
     * Recover the session.
     * <p/>
     * This will cause all unacknowledged messages to be redelivered.
     *
     * @throws JMSException if the session can't be recovered
     */
    public synchronized void recover() throws JMSException {
        stop();             // stop message delivery
        try {
            _database.begin();
            _sent.clear();  // clear the messages in the sent message cache
            _database.commit();
        } catch (Exception exception) {
            rethrow(exception.getMessage(), exception);
        }
        start();           // restart message delivery
    }

    /**
     * Commit the sesion.
     * <p/>
     * This will acknowledge all sent messages for all consumers.
     *
     * @throws JMSException if the session fails to commit
     */
    public synchronized void commit() throws JMSException {
        try {
            _database.begin();
            _sent.acknowledgeAll();
            _database.commit();
        } catch (OutOfMemoryError exception) {
            rethrow("Failed to commit session due to out-of-memory error",
                    exception);
        } catch (Exception exception) {
            rethrow(exception.getMessage(), exception);
        }
    }

    /**
     * Rollback the session.
     * <p/>
     * This will cause all unacknowledged messages to be redelivered.
     *
     * @throws JMSException for any error
     */
    public synchronized void rollback() throws JMSException {
        stop();             // stop message delivery
        try {
            _database.begin();
            _sent.clear();  // clear the messages in the sent message cache
            _database.commit();
        } catch (Exception exception) {
            rethrow(exception.getMessage(), exception);
        }
        start();           // restart message delivery
    }

    /**
     * Return the next available mesage to the specified consumer.
     * <p/>
     * This method is non-blocking. If no messages are available, it will return
     * immediately.
     *
     * @param consumerId the consumer identifier
     * @return the next message or <code>null</code> if none is available
     * @throws JMSException for any JMS error
     */
    public MessageImpl receiveNoWait(long consumerId) throws JMSException {
        MessageImpl result = null;
        if (!_stop.get()) {
            result = doReceive(consumerId, null);
        }
        return result;
    }

    /**
     * Return the next available message to the specified consumer.
     * <p/>
     * This method is non-blocking. However, clients can specify a
     * <code>wait</code> interval to indicate how long they are prepared to wait
     * for a message. If no message is available, and the client indicates that
     * it will wait, it will be notified via the registered {@link
     * JmsMessageListener} if one subsequently becomes available.
     *
     * @param consumerId the consumer identifier
     * @param wait       number of milliseconds to wait. A value of <code>0
     *                   </code> indicates to wait indefinitely
     * @return the next message or <code>null</code> if none is available
     * @throws JMSException for any JMS error
     */
    public MessageImpl receive(long consumerId, long wait) throws JMSException {
        MessageImpl result = null;
        Condition condition;
        if (wait > 0) {
            condition = TimedCondition.before(wait);
        } else {
            condition = new Flag(true);
        }
        if (!_stop.get()) {
            result = doReceive(consumerId, condition);
        } else {
            ConsumerEndpoint consumer = getConsumer(consumerId);
            consumer.setWaitingForMessage(condition);
        }
        return result;
    }