asyncmessageconsumer.java

J2EE开发与Weblogic一书中的源代码

package com.learnweblogic.examples.ch7.topic;

import com.learnweblogic.examples.BaseClient;

import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageListener;
import javax.jms.Session;
import javax.jms.TextMessage;
import javax.jms.Topic;
import javax.jms.TopicConnection;
import javax.jms.TopicConnectionFactory;
import javax.jms.TopicSession;
import javax.jms.TopicSubscriber;
import javax.naming.Context;
import javax.naming.NamingException;

public final class AsyncMessageConsumer
    extends BaseClient
    implements MessageListener {

    private int EXPECTED_MESSAGE_COUNT = 2;
    private int messageCount = 0;

    private TopicSubscriber subscriber;
    private TextMessage msg;

    public AsyncMessageConsumer(String[] argv)
        throws JMSException, NamingException {
        super(argv);

        Context ctx = getInitialContext();

        TopicConnectionFactory tConFactory =
            (TopicConnectionFactory) ctx.lookup(
                "weblogic.jms.ConnectionFactory");

        Topic messageTopic = (Topic) ctx.lookup("MessageTopic");

        TopicConnection tCon = tConFactory.createTopicConnection();

        TopicSession session =
            tCon.createTopicSession(false, /* not a transacted session */
        Session.AUTO_ACKNOWLEDGE);

        subscriber = session.createSubscriber(messageTopic);

        subscriber.setMessageListener(this);

        tCon.start();

    }

    public boolean expectMoreMessages() {
        return messageCount < EXPECTED_MESSAGE_COUNT;
    }

    public void onMessage(Message m) {

        try {
            TextMessage msg = (TextMessage) m;

            System.err.println("Received: " + msg.getText());

        } catch (JMSException e) {
            e.printStackTrace();
        }

        messageCount++;
    }

    public static void main(String[] argv) throws Exception {
        int MAX_TRIES = 10;
        int tryCount = 0;

        AsyncMessageConsumer consumer = new AsyncMessageConsumer(argv);

        while (consumer.expectMoreMessages() && (tryCount < MAX_TRIES)) {
            try {
                Thread.sleep(1000);
            } catch (InterruptedException ignore) {
            }

            tryCount++;
        }
    }

}