httpjmssessionstub.java

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

        return messages;
    }

    /**
     * Create a new queue
     *
     * @param topic the queue to create
     * @throws JMSException if the queue can't be created
     */
    public void createQueue(JmsQueue queue) throws JMSException {
        Vector v = pack("createQueue", 1);
        v.add(queue);
        synchronized (_connection) {
            send(v, true);
            checkReply("createQueue");
        }
    }

    /**
     * Create a new topic
     *
     * @param topic the topic to create
     * @throws JMSException if the topic can't be created
     */
    public void createTopic(JmsTopic topic) throws JMSException {
        Vector v = pack("createTopic", 1);
        v.add(topic);
        synchronized (_connection) {
            send(v, true);
            checkReply("createTopic");
        }
    }

    /**
     * Create a receiver. Get the IP address of the machine the consumer runs
     * on, and the port it is listening to, and pass this to the server, so
     * it can make a new dedicated connection for sending all messages to
     * this client.
     *
     * @param queue the queue to listen to
     * @param clientId the session allocated identifier
     * @param selector the selector to filter messages (may be null)
     * @throws JMSException if the receiver cannot be created
     */
    public void createReceiver(JmsQueue queue, long clientId, String selector)
        throws JMSException {

        startReceiver();
        Vector v = pack("createReceiver", 6);
        v.add(queue);
        v.add(new Long(clientId));
        v.add(selector);
        v.add(_host);
        v.add(String.valueOf(_msgReceiver.getPort()));
        v.add(_url);
        synchronized (_connection) {
            send(v, true);
            checkReply("createReceiver");
        }
    }

    /**
     * Create a queue sender
     *
     * @param queue the queue to send messages to
     * @throws JMSException if the sender cannot be created
     */
    public void createSender(JmsQueue queue) throws JMSException {
        Vector v = pack("createSender", 1);
        v.add(queue);
        synchronized (_connection) {
            send(v, true);
            checkReply("createSender");
        }
    }

    /**
     * Create a queue browser for this session. This allows clients to browse
     * a queue without removing any messages.
     *
     * @param queue the queue to browse
     * @param clientId the identity of the client
     * @param selector the message selector. May be null
     * @throws JMSException if the browser can't be created
     */
    public void createBrowser(JmsQueue queue, long clientId, String selector)
        throws JMSException {

        startReceiver();
        Vector v = pack("createBrowser", 6);
        v.add(queue);
        v.add(new Long(clientId));
        v.add(selector);
        v.add(_host);
        v.add(String.valueOf(_msgReceiver.getPort()));
        v.add(_url);
        synchronized (_connection) {
            send(v, true);
            checkReply("createBrowser");
        }
    }

    /**
     * Delete the receiver for this queue.
     *
     * @param clientId the id of the client to delete
     * @throws JMSException if the receiver cannot be deleted
     */
    public void deleteReceiver(long clientId) throws JMSException {
        Vector v = pack("deleteReceiver", 1);
        v.add(new Long(clientId));
        synchronized (_connection) {
            send(v, true);
            checkReply("deleteReceiver");
        }
    }

    /**
     * Delete the queue browser associated with the specified queue from
     * the session.
     *
     * @param clientId the identity of the browser
     * @throws JMSException if the browser cannot be deleted
     */
    public void deleteBrowser(long clientId) throws JMSException {
        Vector v = pack("deleteBrowser", 1);
        v.add(new Long(clientId));
        synchronized (_connection) {
            send(v, true);
            checkReply("deleteBrowser");
        }
    }

    /**
     * Create a new topic subscriber
     *
     * @param topic the topic to subscribe to
     * @param name the subscribers name
     * @param client the client identity
     * @param selector the selector to filter messages (may be null)
     * @throws JMSException if the topic subscriber can't be created
     */
    public void createSubscriber(JmsTopic topic, String name, long clientId,
                                 String selector, boolean noLocal)
        throws JMSException {

        startReceiver();
        Vector v = pack("createSubscriber", 8);
        v.add(topic);
        v.add(name);
        v.add(new Long(clientId));
        v.add(selector);
        v.add(new Boolean(noLocal));
        v.add(_host);
        v.add(String.valueOf(_msgReceiver.getPort()));
        v.add(_url);
        synchronized (_connection) {
            send(v, true);
            checkReply("createSubscriber");
        }
    }

    /**
     * Create a new topic publisher
     *
     * @param topic the topic to publish to
     * @throws JMSException if the publisher can't be created
     */
    public void createPublisher(JmsTopic topic) throws JMSException {
        Vector v = pack("createPublisher", 1);
        v.add(topic);
        synchronized (_connection) {
            send(v, true);
            checkReply("createPublisher");
        }
    }

    /**
     * Unsubscribe a durable subscription
     *
     * @param name the name used to identify the subscription
     * @throws 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, true);
            checkReply("unsubscribe");
        }
    }

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

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

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

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

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

    // implementation of JmsSessionStubIfc.rollback
    public void rollback() throws JMSException {
        Vector v = pack("rollback", 0);
        synchronized (_connection) {
            send(v, true);
            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, true);
                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, true);
                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, true);
                checkReply("xa_forget");
            }
        } catch (JMSException exception) {
            // rethrow as a XAException
            throw new XAException("Failed to commit session " +
                exception);
        }
    }

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

        try {
            Vector v = pack("xa_getTransactionTimeout", 0);
            synchronized (_connection) {
                send(v, true);
                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();