ipcjmsserverstub.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 2000-2004 (C) Exoffice Technologies Inc. All Rights Reserved.
 *
 * $Id: IpcJmsServerStub.java,v 1.19 2004/01/20 14:11:36 tanderson Exp $
 */

package org.exolab.jms.client.mipc;

import java.io.IOException;
import java.util.HashSet;
import java.util.Hashtable;
import java.util.Vector;

import javax.jms.ExceptionListener;
import javax.jms.JMSException;
import javax.jms.JMSSecurityException;

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

import org.exolab.core.ipc.IpcIfc;
import org.exolab.core.mipc.DisconnectionEventListener;
import org.exolab.core.mipc.MultiplexConnection;
import org.exolab.core.mipc.MultiplexConnectionIfc;
import org.exolab.core.mipc.ObjectChannel;
import org.exolab.core.mipc.TcpConstants;
import org.exolab.jms.client.JmsConnectionStubIfc;
import org.exolab.jms.client.JmsErrorCodes;
import org.exolab.jms.client.JmsServerStubIfc;


/**
 * This class is repsonsible for opening an ipc connection to the server
 * and creating IpcJmsConnections.
 *
 * @version     $Revision: 1.19 $ $Date: 2004/01/20 14:11:36 $
 * @author      <a href="mailto:[e-mail]">Jim Mourikis</a>
 * @see         org.exolab.jms.client.mipc.IpcJmsConnectionStub
 * @see         org.exolab.core.mipc.DisconnectionEventListener
 **/
public class IpcJmsServerStub
    implements JmsServerStubIfc, DisconnectionEventListener {

    /**
     * The multiplex connection to the server
     */
    private MultiplexConnectionIfc _mc = null;

    /**
     * The server channel
     */
    private IpcIfc _connection = null;

    /**
     * The server host address
     */
    private String _serverAddress = TcpConstants.LOCAL_HOST;

    /**
     * The internal server host address if OpenJMS server is serving both
     * internet connections behind a NAT router and internal connections.
     */
    private String _internalServerAddress = null;

    /**
     * The port number the server is listening to
     */
    private int _port = TcpConstants.DEFAULT_PORT;

    /**
     * The set of open IpcJmsConnectionStub instances
     */
    private HashSet _connections = new HashSet();

    /**
     * The message dispatcher
     */
    private IpcJmsMessageListener _listener;

    /**
     * The exception listener is used to communicate server connection
     * problems
     */
    private ExceptionListener _exceptionListener = null;

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


    /**
     * Default constructor for serialization
     */
    public IpcJmsServerStub() {
    }

    /**
     * The constructor has a collection of environment variables
     * which it uses to construct a connection to the remote
     * server
     *
     * @param           env
     */
    public IpcJmsServerStub(Hashtable env) {
        if (env.containsKey(IpcJmsConstants.IPC_SERVER_HOST)) {
            _serverAddress = (String) env.get(IpcJmsConstants.IPC_SERVER_HOST);
        }
        if (env.containsKey(IpcJmsConstants.IPC_INTERNAL_SERVER_HOST)) {
            _internalServerAddress = (String) env.get
                (IpcJmsConstants.IPC_INTERNAL_SERVER_HOST);
        }

        if (env.containsKey(IpcJmsConstants.IPC_SERVER_PORT)) {
            try {
                _port = Integer.parseInt((String) env.get(
                    IpcJmsConstants.IPC_SERVER_PORT));
            } catch (NumberFormatException exception) {
                // failure to convert exception use the
                // default port
                _log.warn("Server port is not a valid format."
                          + " Defaulting to " + _port);
            }
        }
    }

    /**
     * Instantiate an instance of this stub with the specified host and
     * port numbers
     *
     * @param       host                server host address
     * @param       port                server port number
     */
    public IpcJmsServerStub(String host, int port) {
        if ((host != null) &&
            (host.length() > 0)) {
            _serverAddress = host;
        }

        _port = port;
    }

    /**
     * Create a connection to the JMS Server. A new connectionId will be
     * returned if the connection is successful. Create a new IpcJmsConnection
     * object with the given id's.
     *
     * @param clientId the identity of client
     * @param username the client username
     * @param password the client password
     * @throws JMSException
     */
    public synchronized JmsConnectionStubIfc createConnection(
        String clientId, String username, String password)
        throws JMSException {
        JmsConnectionStubIfc stub = null;

        openConnection();
        try {
            String connectionId;
            Vector v = pack("createConnection", clientId, username, password);
            synchronized (_connection) {
                _connection.send(v);
                connectionId = (String) checkReply("createConnection");
            }

            stub = new IpcJmsConnectionStub(
                this, _connection, clientId, connectionId, _listener);
            _connections.add(stub);
        } catch (JMSException exception) {
            throw exception;
        } catch (Exception exception) {
            JMSException error = new JMSException(
                "Failed to create connection");
            error.setLinkedException(exception);
            throw error;
        }
        return stub;
    }

    /**
     * Set the server host. This is the host that the server runs on
     *
     * @param host the name of the host
     */
    public void setServerAddress(String host) {
        _serverAddress = host;
    }

    /**
     * Returns the server host
     *
     * @return the server host
     */
    public String getServerAddress() {
        return _serverAddress;
    }

    /**
     * Set the server port. This is the port that the server runs on
     *
     * @param port the server port number
     */
    public void setServerPort(int port) {
        _port = port;
    }

    /**
     * Returns the server port
     *
     * @return the server port
     */
    public int getServerPort() {
        return _port;
    }

    // implementation of DisconnectionEventListener.disconnected
    public void disconnected(MultiplexConnectionIfc connection) {
        close();

        // notify the exception listener if one is registered
        if (_exceptionListener != null) {
            JMSException exception = new JMSException(
                "Connection to server terminated",
                JmsErrorCodes.CONNECTION_TO_SERVER_DROPPED);
            _exceptionListener.onException(exception);
        }
    }

    // implementation of JmsServerStubIfc.setExceptionListener
    public void setExceptionListener(ExceptionListener listener) {
        _exceptionListener = listener;
    }

    public synchronized void close() {
        // local clean up
        try {
            if (_connection != null) {
                _connection.close();
                _listener.stop();
                _mc.finish();
            }
        } catch (Exception exception) {
            // just swallow :-)
        } finally {
            _mc = null;
            _connection = null;
            _listener = null;
        }
    }

    public synchronized void closed(IpcJmsConnectionStub connection) {
        _connections.remove(connection);
        if (_connections.isEmpty()) {
            close();
        }
    }

    /**
     * Create a multiplexed connection to the server
     *
     * @param serverAddress the address of the server
     * @param internalServerAddress internal server address. May be null.
     * @param port the port number to use
     * @return the connection to the server
     * @throws IOException if the connection cannot be established
     */
    protected MultiplexConnectionIfc createClientConnection(
        String serverAddress, String internalServerAddress, int port)
        throws Exception {

        MultiplexConnectionIfc result = null;

        try {
            result = new MultiplexConnection(serverAddress, port);
        } catch (IOException exception) {
            if (internalServerAddress != null) {
                result = new MultiplexConnection(internalServerAddress, port);
            } else {
                // just rethrow it.
                throw exception;
            }
        }
        return result;
    }

    /**
     * Check to see if an ipc connection has been created. If not, connect with
     * the given hostname and port number.
     *
     * @throws JMSException if a connection cannot be established
     */
    private synchronized void openConnection() throws JMSException {
        if (_mc == null) {
            try {
                _mc = createClientConnection(
                    _serverAddress, _internalServerAddress, _port);
            } catch (Exception exception) {
                throw new JMSException(
                    "Failed to create connection: " + exception);
            }

            _mc.setDisconnectionEventListener(this);
            ((Thread) _mc).start();

            _connection = new ObjectChannel("server", _mc);
            _listener = new IpcJmsMessageListener(_mc);
        }
    }

    /**
     * A convenience method to check the success of operations which return
     * a true on sucess.
     *
     * @param method The requested server function.
     * @return the result of the call, or <code>null</code> if the call
     * didn't return a result
     * @throws JMSException for any failure.
     */
    private Object checkReply(String method) throws JMSException {
        Object result = null;
        Vector v = null;
        try {
            v = (Vector) _connection.receive();
        } catch (Exception err) {
            // rethrow as a JMSException
            throw new JMSException("Operation " + method + " failed: " + err);
        }

        if (v != null) {
            Boolean b = (Boolean) v.get(0);
            if (!b.booleanValue()) {
                if (v.get(1) instanceof JMSException) {
                    throw (JMSException) v.get(1);
                } else {
                    throw new JMSException("Operation " + method +
                        " failed:\n" + v.get(1));
                }
            }
            result = v.get(1);
        } else {
            throw new JMSException("Unknown connection error for " + method);
        }

        return result;
    }

    /**
     * Pack all the data that is required by the server in a vector.
     * Set the size of the vector to be exactly the right size for efficiency.
     *
     * @param method the function to invoke on the server.
     * @param id the unique client id.
     * @param username the client username
     * @param password the client password
     * @return Vector The vector containing all the data.
     */
    private Vector pack(String method, String id, String username,
                        String password) {
        Vector v = new Vector(5);
        v.add("org.exolab.jms.server.mipc.IpcJmsServerConnection");
        v.add(method);
        v.add(id);
        v.add(username);
        v.add(password);
        return v;
    }

}