queueconsumerendpoint.java

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

/**
 * 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 2001-2003 (C) Exoffice Technologies Inc. All Rights Reserved.
 *
 * $Id: QueueConsumerEndpoint.java,v 1.35 2003/08/17 01:32:24 tanderson Exp $
 *
 * Date         Author  Changes
 * 3/1/2001     jima    Created
 */
package org.exolab.jms.messagemgr;

import java.sql.Connection;

import javax.jms.InvalidSelectorException;
import javax.jms.JMSException;
import javax.jms.Session;
import javax.transaction.TransactionManager;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;

import org.exolab.jms.JMSErrorCodes;
import org.exolab.jms.client.JmsDestination;
import org.exolab.jms.client.JmsQueue;
import org.exolab.jms.message.MessageHandle;
import org.exolab.jms.message.MessageImpl;
import org.exolab.jms.persistence.PersistenceException;
import org.exolab.jms.scheduler.Scheduler;
import org.exolab.jms.server.ClientDisconnectionException;
import org.exolab.jms.server.JmsServerSession;


/**
 * A QueueConsumerEndpoint extends {@link ConsumerEndpoint}. This object
 * shares access to a particular queue with other QueueConsumerEndpoint
 * instances.
 *
 * @version     $Revision: 1.35 $ $Date: 2003/08/17 01:32:24 $
 * @author      <a href="mailto:[e-mail]">Jim Alateras</a>
 */
public class QueueConsumerEndpoint
    extends ConsumerEndpoint {

    /**
     * The destination that this consumer subscribes too
     */
    private QueueDestinationCache _cache = null;

    /**
     * The queue that this endpoint is subscribed to
     */
    private JmsQueue _queue = null;

    /**
     * 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(QueueConsumerEndpoint.class);


    /**
     * Construct a QueueConsumerEndpoint, that extends ConsumerEndpoint and
     * is used to manage both synchronous and asynchronous message delivery
     * for this consumer.
     *
     * @param session - the owning session
     * @param clientId - uniquely identifies the remote client within session
     * @param destination - destination that this object was created for
     * @param selector - the selector attached to the consumer, if any.
     * @param scheduler - used to schedule async message delivery.
     * @exception InvalidSelectorException
     */
    QueueConsumerEndpoint(JmsServerSession session, long clientId,
                          JmsQueue queue, String selector, Scheduler scheduler)
        throws InvalidSelectorException {
        super(session, clientId, selector, scheduler);

        // now we should register the endpoint with the destination cache. If
        // the destination cache does not exist then we should create it.
        // All other methods in this class assume that a non-null cache
        // exists.
        if (queue != null) {
            _queue = queue;
            _cache = (QueueDestinationCache)
                DestinationManager.instance().getDestinationCache(queue);
            if (_cache == null) {
                _cache = (QueueDestinationCache)
                    DestinationManager.instance().createDestinationCache(queue);
            }
            _cache.registerConsumer(this);
        }
    }

    /**
     * Return the number of unsent messages
     *
     * @return the number of unsent messages
     */
    public int getMessageCount() {
        return _cache.getMessageCount();
    }

    /**
     * Deliver messages in the cache to the consumer
     *
     * @return <code>true</code> if the endpoint should be rescheduled
     */
    public boolean deliverMessages() {
        boolean reschedule = true;

        for (int index = 0; index < MAX_MESSAGES; index++) {
            // check if we should exit the loop
            if (stopDelivery()) {
                reschedule = false;
                break;
            }

            MessageHandle handle = null;
            try {
                handle = _cache.getMessage(this);
                
                // if the handle is null then there are no more messages
                // to remove so break
                if (handle == null) {
                    reschedule = false;
                    break;
                }
                
                // set the delivered flag only when the message is
                // processed in the session object. Send the message
                // handle to the listener.
                handle.setClientId(getClientId());
                _listener.onMessage(handle, true);
            } catch (ClientDisconnectionException exception) {
                if (handle != null) {
                    _cache.returnMessage(handle);
                }
                _listener = null;
                _log.error(exception, exception);
            } catch (JMSException exception) {
                if (exception.getErrorCode().equals(
                        JMSErrorCodes.FailedToResolveHandle)) {
                    // do not return message back to the cache
                    _log.error("Dropping handle " + handle
                               + " since we cannot resolve it.");
                } else {
                    _log.error(exception, exception);
                    if (handle != null) {
                        _cache.returnMessage(handle);
                    }
                }
            } catch (Exception exception) {
                if (handle != null) {
                    _cache.returnMessage(handle);
                }
                _log.error(exception);
            }
        } 
        return reschedule;
    }

    // override ConsumerEndpoint.setMessageListener
    public void setMessageListener(InternalMessageListener listener) {
        // if the listener is null we unregister from the destination. If
        // the listener is not null then we register with it. When we
        // unregister we need to determine what happens to any messages
        // in the sent and
        if (listener == null) {
            _cache.unregisterConsumer(this);
        } else {
            _cache.registerConsumer(this);
        }

        super.setMessageListener(listener);
    }

    // implementation of ConsumerEndpoint.receiveMessage
    public MessageHandle receiveMessage(long wait) {
        MessageHandle handle = getMessageFromCache();
        if ((handle == null) &&
            (wait >= 0)) {

            // set a flag indicating that we are waiting
            // for a message
            setWaitingForMessage();

            // perform a double check and we receive a
            // message then clear the previous set flag
            handle = getMessageFromCache();
            if (handle != null) {
                clearWaitingForMessage();
            }
        }

        return handle;
    }

    /**
     * Check whether a listener has been registered with this endpoint to
     * support async message delivery
     *
     * @return boolean - true if it has
     */
    public boolean hasMessageListener() {
        return _listener != null;
    }

    // implementation of ConsumerEndpoint.unregister
    public void unregister() {
        _cache.unregisterConsumer(this);
    }

    // implementation of ConsumerEndpoint.getDestination
    public JmsDestination getDestination() {
        return _queue;
    }

    // override ConsumerEndpoint.messageAdded
    public boolean messageAdded(MessageImpl message) {
        if (_listener != null) {
            schedule();
        } else {
            // check to see whether the consumer is waiting to
            // be notified
            notifyMessageAvailable();
        }

        return true;
    }

    // override ConsumerEndpoint.persistentMessageAdded
    public boolean persistentMessageAdded(Connection connection,
                                          MessageImpl message)
        throws PersistenceException {
        if (_listener != null) {
            schedule();
        } else {
            // check to see whether the consumer is waiting to
            // be notified
            notifyMessageAvailable();
        }

        return true;
    }

    // override ConsumerEndpoint.messageRemoved
    public synchronized boolean messageRemoved(MessageImpl message) {
        return false;
    }

    /**
     * Closes this endpoint
     */
    protected void doClose() {
        // unregister from the DestinationCache
        _cache.unregisterConsumer(this);
    }

    /**
     * Return a message from the corresponding cache for this consumer or
     * null if one is not available
     *
     * @return MessageHandle - the handle or null
     */
    private MessageHandle getMessageFromCache() {
        MessageHandle handle = _cache.getMessage(this);
        if (handle != null) {
            handle.setClientId(getClientId());
        }

        return handle;
    }

} //-- QueueConsumerEndpoint