ipcjmssessionstub.java

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

        v.add(name);
        v.add(new Long(clientId));
        v.add(selector);
        v.add(new Boolean(noLocal));
        v.add(((ObjectChannel) _connection).getConnection().getHost());
        v.add(String.valueOf
            (((ObjectChannel) _connection).getConnection().getPort()));
        v.add("n");
        synchronized (_connection) {
            send(v);
            checkReply("createSubscriber");
        }
    }


    /**
     * Create a new topic publisher
     *
     * @param         topic  The topic to publish to
     * @exception     JMSException On error
     *
     */
    public void createPublisher(JmsTopic topic) throws JMSException {
        Vector v = pack("createPublisher", 1);
        v.add(topic);
        synchronized (_connection) {
            send(v);
            checkReply("createPublisher");
        }
    }


    /**
     * Unsubscribe a durable subscription
     *
     * @param       name                the name used to identify the
     *                                  subscription
     * @exception   JMSException        if the subscription cannot be removed
     */
    public void unsubscribe(String name)
        throws JMSException {
        Vector v = pack("unsubscribe", 1);
        v.add(name);
        synchronized (_connection) {
            send(v);
            checkReply("unsubscribe");
        }
    }

    /**
     * Delete the subscriber for this topic
     *
     * @param         clientId - the client identity
     * @exception     JMSException On error
     *
     */
    public void deleteSubscriber(long clientId) throws JMSException {
        Vector v = pack("deleteSubscriber", 1);
        v.add(new Long(clientId));
        synchronized (_connection) {
            send(v);
            checkReply("deleteSubscriber");
        }
    }

    /**
     * Stop message delivery for this session.
     *
     * @exception     JMSException On error
     *
     */
    public void stopMessageDelivery() throws JMSException {
        Vector v = pack("stopMessageDelivery", 0);
        synchronized (_connection) {
            send(v);
            checkReply("stopMessageDelivery");
        }
    }


    /**
     * Start message delivery for this session.
     *
     * @exception     JMSException On error
     *
     */
    public void startMessageDelivery() throws JMSException {
        Vector v = pack("startMessageDelivery", 0);
        synchronized (_connection) {
            send(v);
            checkReply("startMessageDelivery");
        }
    }

    // implementation of JmsSessionStubIfc.recover
    public void recover() throws JMSException {
        Vector v = pack("recover", 0);
        synchronized (_connection) {
            send(v);
            checkReply("recover");
        }
    }

    // implementation of JmsSessionStubIfc.commit
    public void commit() throws JMSException {
        Vector v = pack("commit", 0);
        synchronized (_connection) {
            send(v);
            checkReply("commit");
        }
    }

    // implementation of JmsSessionStubIfc.rollback
    public void rollback() throws JMSException {
        Vector v = pack("rollback", 0);
        synchronized (_connection) {
            send(v);
            checkReply("rollback");
        }
    }

    // implementation of JmsSessionStubIfc.commit
    public void commit(Xid xid, boolean onePhase)
        throws XAException {
        try {
            Vector v = pack("xa_commit", 2);
            v.add(xid);
            v.add(new Boolean(onePhase));
            synchronized (_connection) {
                send(v);
                checkReply("xa_commit");
            }
        } catch (JMSException exception) {
            // rethrow as a XAException
            throw new XAException("Failed to commit session " +
                exception);
        }
    }

    // implementation of JmsSessionStubIfc.end
    public void end(Xid xid, int flags)
        throws XAException {
        try {
            Vector v = pack("xa_end", 2);
            v.add(xid);
            v.add(new Integer(flags));
            synchronized (_connection) {
                send(v);
                checkReply("xa_end");
            }
        } catch (JMSException exception) {
            // rethrow as a XAException
            throw new XAException("Failed to commit session " +
                exception);
        }
    }

    // implementation of JmsSessionStubIfc.forget
    public void forget(Xid xid)
        throws XAException {
        try {
            Vector v = pack("xa_forget", 1);
            v.add(xid);
            synchronized (_connection) {
                send(v);
                checkReply("xa_forget");
            }
        } catch (JMSException exception) {
            // rethrow as a XAException
            throw new XAException("Failed to commit session " +
                exception);
        }
    }

    // implementation of JmsSessionStubIfc.getResourceManagerId
    public String getResourceManagerId() throws XAException {
        String rid = null;

        try {
            Vector v = pack("xa_getResourceManagerId", 0);
            synchronized (_connection) {
                send(v);
                Vector reply = checkReply("xa_getResourceManagerId");
                Boolean result = (Boolean) reply.get(0);

                // check that the call completed before
                // extracting the message
                if (result.booleanValue()) {
                    rid = (String) reply.get(1);
                }
            }
        } catch (JMSException exception) {
            // rethrow as a XAException
            throw new XAException("Failed to getResourceManagerId session " +
                exception);
        }

        return rid;
    }

    // implementation of JmsSessionStubIfc.getTransactionTimeout
    public int getTransactionTimeout()
        throws XAException {
        int timeout = 0;

        try {
            Vector v = pack("xa_getTransactionTimeout", 0);
            synchronized (_connection) {
                send(v);
                Vector reply = checkReply("xa_getTransactionTimeout");
                Boolean result = (Boolean) reply.get(0);

                // check that the call completed before
                // extracting the message
                if (result.booleanValue()) {
                    timeout = ((Integer) reply.get(1)).intValue();
                }
            }
        } catch (JMSException exception) {
            // rethrow as a XAException
            throw new XAException("Failed to getTransactionTimeout session " +
                exception);
        }

        return timeout;
    }

    // implementation of JmsSessionStubIfc.prepare
    public int prepare(Xid xid)
        throws XAException {
        int value = 0;

        try {
            Vector v = pack("xa_prepare", 1);
            v.add(xid);
            synchronized (_connection) {
                send(v);
                Vector reply = checkReply("xa_prepare");
                Boolean result = (Boolean) reply.get(0);

                // check that the call completed before
                // extracting the message
                if (result.booleanValue()) {
                    value = ((Integer) reply.get(1)).intValue();
                }
            }
        } catch (JMSException exception) {
            // rethrow as a XAException
            throw new XAException("Failed to prepare session " +
                exception);
        }

        return value;
    }

    // implementation of JmsSessionStubIfc.recover
    public Xid[] recover(int flag)
        throws XAException {
        Xid[] xids = new Xid[0];

        try {
            Vector v = pack("xa_recover", 1);
            v.add(new Integer(flag));
            synchronized (_connection) {
                send(v);
                Vector reply = checkReply("xa_recover");
                Boolean result = (Boolean) reply.get(0);

                // check that the call completed before
                // extracting the message
                if (result.booleanValue()) {
                    xids = (Xid[]) reply.get(1);
                }
            }
        } catch (JMSException exception) {
            // rethrow as a XAException
            throw new XAException("Failed to recover session " +
                exception);
        }

        return xids;
    }

    // implementation of JmsSessionStubIfc.rollback
    public void rollback(Xid xid)
        throws XAException {
        try {
            Vector v = pack("xa_rollback", 1);
            v.add(xid);
            synchronized (_connection) {
                send(v);
                checkReply("xa_rollback");
            }
        } catch (JMSException exception) {
            // rethrow as a XAException
            throw new XAException("Failed to rollback session " +
                exception);
        }
    }

    // implementation of JmsSessionStubIfc.setTransactionTimeout
    public boolean setTransactionTimeout(int seconds)
        throws XAException {
        boolean value = false;

        try {
            Vector v = pack("xa_setTransactionTimeout", 1);
            v.add(new Integer(seconds));
            synchronized (_connection) {
                send(v);
                Vector reply = checkReply("xa_setTransactionTimeout");
                Boolean result = (Boolean) reply.get(0);

                // check that the call completed before
                // extracting the message
                if (result.booleanValue()) {
                    value = ((Boolean) reply.get(1)).booleanValue();
                }
            }
        } catch (JMSException exception) {
            // rethrow as a XAException
            throw new XAException("Failed to setTransactionTimeout " +
                exception);
        }

        return value;
    }

    // implementation of JmsSessionStubIfc.start
    public void start(Xid xid, int flags)
        throws XAException {
        try {
            Vector v = pack("xa_start", 2);
            v.add(xid);
            v.add(new Integer(flags));
            synchronized (_connection) {
                send(v);
                checkReply("xa_start");
            }
        } catch (JMSException exception) {
            // rethrow as a XAException
            throw new XAException("Failed to start session " +
                exception);
        }
    }

    /**
     * Set a message listener to be called when new Messages arrive from the
     * server.
     *
     * @param listener A reference to the client listener.
     */
    public void setMessageListener(JmsMessageListener listener) {
        _listener.setListener(_sessionId, listener);
    }

    // implementation of JmsSessionStubIfc.enableAsynchronousDelivery
    public void enableAsynchronousDelivery(long clientId, String id,
                                           boolean enable)
        throws JMSException {
        Vector v = pack("enableAsynchronousDelivery", 3);
        v.add(new Long(clientId));
        v.add(id);
        v.add(new Boolean(enable));
        synchronized (_connection) {
            send(v);
            checkReply("enableAsynchronousDelivery");
        }
    }

    /**
     * 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 activate on the server.
     * @param numParams The number of paramaters this method will require.
     * @return Vector The vector containing all the data.
     *
     */
    private Vector pack(String method, int numParams) {
        Vector v = new Vector(5 + numParams);
        v.add("org.exolab.jms.server.mipc.IpcJmsSessionConnection");
        v.add(method);
        v.add(_clientId);
        v.add(_connectionId);
        v.add(_sessionId);
        return v;
    }

    /**
     * A convenience method to check the success of operations which return
     * a true on sucess.
     *
     * @param method The requested server function.
     * @exception JMSException On any failure.
     *
     */
    private Vector checkReply(String method) throws JMSException {
        Vector v = null;
        try {
            v = (Vector) _connection.receive();
        } catch (Exception err) {
            // rethrow as a JMSException
            throw new JMSException("Operation " + method + " failed:\n" + 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));
                }
            }
        } else {
            throw new JMSException("Unknown connection error for " + method);
        }

        return v;
    }

    /**
     * A convenience method to send a packed command to the server.
     *
     * @throws JMSException for any failure.
     */
    private void send(Vector v) throws JMSException {
        try {
            _connection.send(v);
        } catch (Exception err) {
            // rethrow as a JMSException
            throw new JMSException("Operation Failed" + err);
        }
    }

} //-- IpcJmsSessionStub