ipcjmssessionconnection.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-2003 (C) Exoffice Technologies Inc. All Rights Reserved.
 *
 * $Id: IpcJmsSessionConnection.java,v 1.28 2003/08/30 07:58:51 tanderson Exp $
 *
 * Date         Author  Changes
 * $Date	    jimm    Created
 */
package org.exolab.jms.server.mipc;

import java.io.Serializable;
import java.util.Vector;

import javax.jms.JMSException;
import javax.jms.Message;
import javax.transaction.xa.XAException;
import javax.transaction.xa.XAResource;
import javax.transaction.xa.Xid;

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

import org.exolab.core.ipc.NotifierIfc;
import org.exolab.core.mipc.MultiplexConnectionIfc;
import org.exolab.jms.client.JmsQueue;
import org.exolab.jms.client.JmsTopic;
import org.exolab.jms.message.MessageImpl;
import org.exolab.jms.server.JmsServerConnection;
import org.exolab.jms.server.JmsServerConnectionManager;
import org.exolab.jms.server.JmsServerSession;


/**
 * This is the server side receiver for JmsSession requests. All requests are
 * unpacked and passed on to the appropriate JmsServerSession object.
 *
 * @version     $Revision: 1.28 $ $Date: 2003/08/30 07:58:51 $
 * @author      <a href="mailto:[e-mail]">Jim Mourikis</a>
 * @see         org.exolab.jms.server.mipc.IpcJmsReceiver
 * @see	       	org.exolab.jms.server.JmsServerConnection
 * @see	       	org.exolab.jms.server.JmsServerConnectionManager
 * @see         org.exolab.jms.server.JmsServerSession
 */
public class IpcJmsSessionConnection implements NotifierIfc {

    /**
     * The server
     */
    private IpcJmsServer _server;

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


    /**
     * Construct a new <code>IpcJmsSessionConnection</code>
     *
     * @param server the server instance
     */
    public IpcJmsSessionConnection(IpcJmsServer server) {
        _server = server;
    }

    /**
     * A new request has been received.
     * Carry out the request, and pass back any relevent data.
     *
     * @param ob The data received
     * @param id The unique IPC id of the connection
     * @param The unique identifier of this connection.
     * @return Object Return any requested result. This must never be null.
     *
     */
    public Serializable notify(Object ob, String id) {
        Vector v = (Vector) ob;
        String func = (String) v.get(1);
        JmsServerSession session = getSession(id, v);
        Serializable result = null;
        if (session == null) {
            JMSException error = new JMSException(
                "Failed to process request " + func
                + ": session not found");
            result = pack(Boolean.FALSE, error);
        } else {
            if (func.equals("close")) {
                result = close(session, getConnection(id));
            } else if (func.equals("acknowledgeMessage")) {
                result = acknowledgeMessage(session, (Long) v.get(5),
                                            (String) v.get(6));
            } else if (func.equals("sendMessage")) {
                result = sendMessage(session, (Message) v.get(5));
            } else if (func.equals("sendMessages")) {
                result = sendMessages(session, (Vector) v.get(5));
            } else if (func.equals("receiveMessage")) {
                result = receiveMessage(session, (Long) v.get(5),
                                        (Long) v.get(6));
            } else if (func.equals("receiveMessages")) {
                result = receiveMessages(session, (Long) v.get(5),
                                         (Integer) v.get(6));
            } else if (func.equals("createQueue")) {
                result = createQueue(session, (JmsQueue) v.get(5));
            } else if (func.equals("createTopic")) {
                result = createTopic(session, (JmsTopic) v.get(5));
            } else if (func.equals("createReceiver")) {
                result = createReceiver(
                    session, (JmsQueue) v.get(5), (Long) v.get(6),
                    (String) v.get(7), getConnection(id), (String) v.get(8),
                    (String) v.get(9), (String) v.get(10));
            } else if (func.equals("createSender")) {
                result = createSender(session, (JmsQueue) v.get(5));
            } else if (func.equals("createBrowser")) {
                result = createBrowser(
                    session, (JmsQueue) v.get(5), (Long) v.get(6),
                    (String) v.get(7), getConnection(id), (String) v.get(8),
                    (String) v.get(9), (String) v.get(10));
            } else if (func.equals("deleteReceiver")) {
                result = deleteReceiver(session, (Long) v.get(5));
            } else if (func.equals("deleteSender")) {
                result = deleteSender(session, (Long) v.get(5));
            } else if (func.equals("deleteBrowser")) {
                result = deleteBrowser(session, (Long) v.get(5));
            } else if (func.equals("createSubscriber")) {
                result = createSubscriber(
                    session, (JmsTopic) v.get(5), (String) v.get(6),
                    (Long) v.get(7), (String) v.get(8), (Boolean) v.get(9),
                    getConnection(id), (String) v.get(10), (String) v.get(11),
                    (String) v.get(12));
            } else if (func.equals("createPublisher")) {
                result = createPublisher(session, (JmsTopic) v.get(5));
            } else if (func.equals("deleteSubscriber")) {
                result = deleteSubscriber(session, (Long) v.get(5));
            } else if (func.equals("deletePublisher")) {
                result = deletePublisher(session, (JmsTopic) v.get(5));
            } else if (func.equals("unsubscribe")) {
                result = unsubscribe(session, (String) v.get(5));
            } else if (func.equals("stopMessageDelivery")) {
                result = stopMessageDelivery(session);
            } else if (func.equals("startMessageDelivery")) {
                result = startMessageDelivery(session);
            } else if (func.equals("recover")) {
                result = recover(session);
            } else if (func.equals("commit")) {
                result = commit(session);
            } else if (func.equals("rollback")) {
                result = rollback(session);
            } else if (func.equals("xa_commit")) {
                result = XACommit(session, (Xid) v.get(5), (Boolean) v.get(6));
            } else if (func.equals("xa_end")) {
                result = XAEnd(session, (Xid) v.get(5), (Integer) v.get(6));
            } else if (func.equals("xa_forget")) {
                result = XAForget(session, (Xid) v.get(5));
            } else if (func.equals("xa_getTransactionTimeout")) {
                result = XAGetTransactionTimeout(session);
            } else if (func.equals("xa_recover")) {
                result = XARecover(session, (Integer) v.get(5));
            } else if (func.equals("xa_rollback")) {
                result = XARollback(session, (Xid) v.get(5));
            } else if (func.equals("xa_setTransactionTimeout")) {
                result = XASetTransactionTimeout(session, (Integer) v.get(5));
            } else if (func.equals("xa_start")) {
                result = XAStart(session, (Xid) v.get(5), (Integer) v.get(6));
            } else if (func.equals("xa_prepare")) {
                result = XAPrepare(session, (Xid) v.get(5));
            } else if (func.equals("xa_getResourceManagerId")) {
                result = XAGetResourceManagerId(session);
            } else if (func.equals("enableAsynchronousDelivery")) {
                result = enableAsynchronousDelivery(
                    session, (Long) v.get(5), (String) v.get(6),
                    (Boolean) v.get(7));
            } else {
                JMSException error = new JMSException(
                    "Unknown request received: " + func);
                result = pack(Boolean.FALSE, error);
            }
        }

        return result;
    }

    /**
     * The connection has been broken.              
     *
     * @param id The unique identifier of this connection.
     */
    public void disconnection(String id) {
    }

    /**
     * A convenience routine to get the session.
     *
     * @param id The unique IPC id of the connection
     * @param v The vector containing the packed data from the client.
     * @return JmsServerSession The required session or null if not found.
     *
     */
    protected JmsServerSession getSession(String id, Vector v) {
        // Lookup the connection using the unique connection id
        JmsServerConnection connection =
            JmsServerConnectionManager.instance().getConnection
            ((String) v.get(2));

        JmsServerSession session = null;

        if (connection != null) {
            session = connection.getSession((String) v.get(4));
        } else if (_log.isDebugEnabled()) {
            _log.debug("Failed to locate connection=" + v.get(2));
        }
        if (session == null && _log.isDebugEnabled()) {
            _log.debug("Failed to locate session=" + v.get(4));
        }

        return session;
    }

    /**
     * A convenience routine to get the MutliplexConnection.
     *
     * @param id The unique IPC id of the connection
     * @return MultiplexConnection The connection associated with the ipc channel.
     *
     */
    protected MultiplexConnectionIfc getConnection(String id) {
        IpcServerChannel channel = IpcServerChannel.getServerChannel(id);
        MultiplexConnectionIfc mc = null;
        if (channel != null) {
            mc = channel.getConnection();
        }
        return mc;
    }

    /**
     * A close session request has been received.
     *
     * @param session The session to close
     * @return Vector The result of the request.
     *
     */
    protected Vector close(JmsServerSession session,
                           MultiplexConnectionIfc connection) {
        try {
            _server.removeConnection(session, connection);
            session.close();
        } catch (JMSException exception) {
            return pack(Boolean.FALSE, exception);
        }

        return pack(Boolean.TRUE, null);
    }

    /**
     * Acknowledge a JmsMessage
     *
     * @param session - the session the request is for
     * @param clientId - the identity of the client that sent it
     * @param id The id of the message
     * @return Vector The result of the request.
     *
     */
    protected Vector acknowledgeMessage(JmsServerSession session,
                                        Long clientId, String id) {
        try {
            session.acknowledgeMessage(clientId.longValue(), id);
        } catch (JMSException exception) {
            return pack(Boolean.FALSE, exception);
        }

        return pack(Boolean.TRUE, null);
    }

    /**
     * A JmsMessage has been sent.
     *
     * @param session The session the request is for.
     * @param message The message to process.
     * @return Vector The result of the request.
     *
     */
    protected Vector sendMessage(JmsServerSession session, Message message) {
        boolean persistent = ((MessageImpl) message).isPersistent();

        try {
            session.sendMessage(message);
        } catch (JMSException exception) {