ipcserverchannel.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: IpcServerChannel.java,v 1.8 2003/08/07 13:33:09 tanderson Exp $
 *
 * Date         Author  Changes
 * 2/14/2001    artw    Created
 */

package org.exolab.jms.server.mipc;

import java.io.Serializable;
import java.util.HashMap;

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

import org.exolab.core.ipc.NotifierIfc;
import org.exolab.core.mipc.DisconnectionEventListener;
import org.exolab.core.mipc.MultiplexConnectionIfc;
import org.exolab.core.mipc.ObjectChannel;


/**
 * An instance of this class represents a client's connection to the server.
 * It encapsulates the MultiplexConnection and is responsible for listening
 * to ipc messages and dispatching them to specified NotifierIfc client.
 *
 * This object derives from Thread and runs in its own thread.  shutdown()
 * should be used to gracefully stop this thread.
 *
 * @version     $Revision: 1.8 $
 * @author      <a href="mailto:[e-mail]">Art Whitten</a>
 * @see         org.exolab.core.mipc.MultiplexConnection
 **/
public class IpcServerChannel
    extends Thread
    implements DisconnectionEventListener {

    /**
     * The associated with the client connection
     */
    private MultiplexConnectionIfc _connection;

    /**
     * Object channel used to receive remote ipc commands.
     */
    private ObjectChannel _oc;

    /**
     *The client that wants to be notified whenever a ipc message is received.
     */
    private NotifierIfc _client;

    /**
     * Unique ID for this instance
     */
    private String _id;

    /**
     * Used for stopping the Thread
     */
    private volatile boolean _stopRequested = false;

    /**
     * Used for creating a unique id
     */
    private static int _nextID = 0;

    /**
     * Keeps track of all of the IpcServerChannels.  The key is the
     * _id and the value is the object.
     */
    private static HashMap _channels = new HashMap();

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


    /**
     * Creates new IpcServerChannel on the specified connection.
     *
     * @param connection The MultiplexConnection to listen for ipc commands on.
     * @param client This object gets notified of all ipc messages.
     * @param channel The name of the channel to listen on.
     */
    public IpcServerChannel(MultiplexConnectionIfc connection,
                            NotifierIfc client, String channel) {
        super(channel + "-" + getNextID());
        _connection = connection;
        _connection.setDisconnectionEventListener(this);
        _client = client;
        _oc = new ObjectChannel(channel, connection);

        // The name that we created for the thread is unique.
        _id = getName();

        // Keep track of all the active server channels by unique id
        addServerChannel(this);
    }

    /**
     * Return the underlying MultiplexConnection
     *
     * @return the underlying connection
     */
    public MultiplexConnectionIfc getConnection() {
        return _connection;
    }

    /**
     * This method is called when the thread starts.  It sits in a loop
     * receiving ipc messages.  When a message is received, it is passed onto
     * the client notifier.
     */
    public void run() {
        while (!_stopRequested) {
            try {
                Object object = _oc.receive();
                Serializable returnValue = _client.notify(object, _id);
                if (returnValue != null) {
                    _oc.send(returnValue);
                }
            } catch (Exception exception) {
                if (!_stopRequested) {
                    _log.debug(exception, exception);
                }
            }
        }

        try {
            _client.disconnection(_id);
            _oc.close();
        } catch (Exception ignore) {
            // eat exceptions, we don't care here
        } finally {
            removeServerChannel(this);
            _client = null;
            _connection = null;
        }
    }

    /**
     * Stop execution of this thread.
     */
    public void shutdown() {
        _stopRequested = true;
        interrupt();
    }

    // implementation of DisconnectionEventListener.diconnected
    public void disconnected(MultiplexConnectionIfc connection) {
        shutdown();
        new IpcJmsServerConnection().disconnection(_id);
    }

    /**
     * Given a IpcServerChannel uniqueID string, return the associated
     * instance.
     *
     * @param id The string ID of the IpcServerChannel we are looking for.
     * @return IpcServerChannel The found object, null if none.
     */
    public static synchronized IpcServerChannel getServerChannel(String id) {
        return (IpcServerChannel) _channels.get(id);
    }

    /**
     * Add a new IpcServerChannel to the list of connections so that clients
     * can get to it given its uniqueID
     *
     * @param newChannel The IpcServerChannel to add.
     */
    private synchronized static void addServerChannel(
        IpcServerChannel newChannel) {
        _channels.put(newChannel._id, newChannel);
    }

    /**
     * Remove a IpcServerChannel from the list of active channels
     *
     * @param removeChannel The IpcServerChannel to remove.
     */
    private synchronized static void removeServerChannel(
        IpcServerChannel removeChannel) {
        _channels.remove(removeChannel._id);
    }

    private synchronized static int getNextID() {
        return _nextID++;
    }

}