messageconsumer.java

javaP2P技术内幕课程111213141516源代码

import javax.swing.*;
import javax.jms.*;
import javax.naming.*;
import java.io.*;
import java.util.*;


public class MessageConsumer {
    //We would like to keep a reference of TopicConnection
    //and TopicSession so that we can destroy it at the end.
    private TopicConnection topicConnection = null;
    private TopicSession topicSession = null;

    //This Subscriber will listen for incoming messages.
    private TopicSubscriber topicSubscriber = null;

    //This listener will receive the incoming messages.
    private TextListener topicListener = null;

    //The TextArea to disply the messages.
    private JTextArea textArea = null;

    //Names of machines over the network.
    //These machines will have JMS Providers running.
    //In real applications,names of machines should come
    //from the GUI.
    private final String COMPUTER1 = "computer6";
    private final String COMPUTER2 = "computer13";
    private final String COMPUTER3 = "computer14";
    private final String COMPUTER4 = "computer15";

    //Connect to the JMS Provider and start listening.
    public void connectToProvider(String topicName, String jmsProvider) {
        //We will get a reference of System properties.
        //We will add the Properties related to JMS functionality.
        Properties env = System.getProperties();
        env.put("com.sun.jms.internal.java.naming.factory.initial", 
                "com.sun.enterprise.naming.SerialInitContextFactory");


        //Port number where the Naming service of
        //remote JMS Provider is listening.
        env.put("rg.omg.CORBA.ORBInitialPort", "1050 ");

        try {
            //Set the network address of target machine
            //where JMS Provider is running.
            if (jmsProvider.equals("Server1 ")) {
                env.put("org.omg.CORBA.ORBInitialHost", COMPUTER1);
            } else if (jmsProvider.equals("Server2")) {
                env.put("org.omg.CORBA.ORBInitialHost", COMPUTER2);
            } else if (jmsProvider.equals("Server3")) {
                env.put("org.omg.CORBA.ORBInitialHost", COMPUTER3);
            } else if (jmsProvider.equals("Server4")) {
                env.put("org.omg.CORBA.ORBInitialHost", COMPUTER4);
            }

            //Get an object of JNDI InitialContext
            //from given JMS Provider.
            Context jndiContext = new InitialContext(env);

            //Get JMS Administrator Objects
            TopicConnectionFactory topicConnectionFactory = 
                    (TopicConnectionFactory) jndiContext.lookup(
                            "TopicConnectionFactory");
            Topic topic = (Topic) jndiContext.lookup(topicName);


            //Establish the connection.
            topicConnection = topicConnectionFactory.createTopicConnection();


            //Create the session.
            topicSession = topicConnection.createTopicSession(false, 
                                                              Session.AUTO_ACKNOWLEDGE);


            //Create the subscriber for the topic of interest.
            //Specify a TextListener object (inner class)
            //who will receive messages.
            topicSubscriber = topicSession.createSubscriber(topic);
            topicListener = new TextListener();
            topicSubscriber.setMessageListener(topicListener);
            textArea.setText("starting the listener for messages");


            //We are all set,Start listening.
            topicConnection.start();
        } catch (NamingException e) {
            textArea.append("JNDI lookup failed: " + e.toString());
        } catch (JMSException e) {
            textArea.append("JMS Exception occurred: " + e.toString());
        }
    } //connectToProvider()

    //Stop Listening for messages and close the Connection.
    public void stopListening() {
        if (topicConnection != null) {
            try {
                topicSubscriber.close();
                topicSession.close();
                topicConnection.close();
                textArea.setText("\nStoped Listining");
            } catch (JMSException e) {
                textArea.setText("\nException in listing");
            }
        }
    } //stopListening()

    public void setDisplayArea(JTextArea textArea) {
        this.textArea = textArea;
    } //setDisplayArea

    //Inner class to receive messages.
    private class TextListener implements MessageListener {
        public void onMessage(Message message) {
            TextMessage msg = null;

            try {
                //Our application can only handle Text Messages.
                if (message instanceof TextMessage) {
                    msg = (TextMessage) message;
                    textArea.append("\n \n The Message is Received:\n\t " + 
                                    msg.getText());
                } else {
                    textArea.append("\n Message of wrong type: " + 
                                    message.getClass().getName());
                } //else
            } catch (JMSException e) {
                textArea.append("JMSException in onMessage(): " + 
                                e.toString());
            } catch (Throwable te) {
                textArea.append("Exception in onMessage(): " + 
                                te.getMessage());
            } //catch
        } //onMessage()
    } //TextListener class
} //Message Consumer class