topicjms.java

本书中的源代码是以JBuilder工程形式组织的

package mytopicjms;

import javax.jms.*;
import javax.naming.*;
import java.util.Hashtable;

/**
 * <p>Title: </p>
 * <p>Description: </p>
 * <p>Copyright: Copyright (c) 2000</p>
 * <p>Company: </p>
 * @author not attributable
 * @version 1.0
 *
 * Use this class to publish and subscribe to messages.
 * To send a text message:
 * <code>
 * TopicJMS topicJMS = new TopicJMS();
 * topicJMS.setEnvironment(hashtable);  //Specify any vendor-specific JNDI settings here
 * topicJMS.publishText("Hello world");
 * topicJMS.close(); //Release resources
 * </code>
 *
 * <code>
 * To receive a message:
 * TopicJMS topicJMS = new TopicJMS();
 * topicJMS.getTopicSubscriber();
 * </code>
 */

public class TopicJMS implements MessageListener {
  private static Context context = null;
  private boolean transacted = false;
  private int acknowledgementMode = Session.AUTO_ACKNOWLEDGE;
  private Hashtable environment = null;
  private TopicConnectionFactory topicConnectionFactory = null;
  private TopicConnection topicConnection = null;
  private TopicSession topicSession = null;
  private TopicPublisher topicPublisher = null;
  private TopicSubscriber topicSubscriber = null;
  private Topic topic = null;
  private String topicConnectionFactoryName = "weblogic.jms.ConnectionFactory";
  private String topicName = "MyJMSTopic";
  private String clientId = "";
  private String durableName = "";
  private boolean durable = false;
  public boolean isTransacted() {
    return transacted;
  }
  public void setTransacted(boolean transacted) {
    this.transacted = transacted;
  }
  public int getAcknowledgementMode() {
    return acknowledgementMode;
  }
  public void setAcknowledgementMode(int acknowledgementMode) {
    this.acknowledgementMode = acknowledgementMode;
  }
  public Hashtable getEnvironment() {
    return environment;
  }
  public void setEnvironment(Hashtable environment) {
    this.environment = environment;
  }
  public String getTopicConnectionFactoryName() {
    return topicConnectionFactoryName;
  }
  public void setTopicConnectionFactoryName(String topicConnectionFactoryName) {
    this.topicConnectionFactoryName = topicConnectionFactoryName;
  }
  public String getTopicName() {
    return topicName;
  }
  public void setTopicName(String topicName) {
    this.topicName = topicName;
  }
  public String getClientId() {
    return clientId;
  }
  public void setClientId(String clientId) {
    this.clientId = clientId;
  }
  public String getDurableName() {
    return durableName;
  }
  public void setDurableName(String durableName) {
    this.durableName = durableName;
  }
  public boolean isDurable() {
    return durable;
  }
  public void setDurable(boolean durable) {
    this.durable = durable;
  }
  Context getContext() throws Exception {
    if (context == null) {
      try {
        context = new InitialContext(environment);
      }
      catch(Exception ex) {
        ex.printStackTrace();
        throw ex;
      }
    }
    return context;
  }
  public TopicConnectionFactory getTopicConnectionFactory() throws Exception {
    if (topicConnectionFactory == null) {
      Object obj = getContext().lookup(topicConnectionFactoryName);
      topicConnectionFactory = (TopicConnectionFactory) obj;
    }
    return topicConnectionFactory;
  }
  public TopicConnection getTopicConnection(boolean consumer) throws Exception {
    if (topicConnection == null) {
      topicConnection = getTopicConnectionFactory().createTopicConnection();
      topicConnection.start();
      if (isDurable() && consumer) {
        topicConnection.setClientID(clientId);
      }
    }
    return topicConnection;
  }
  public TopicSession getTopicSession(boolean consumer) throws Exception {
    if (topicSession == null) {
      topicSession = getTopicConnection(consumer).createTopicSession(isTransacted(), getAcknowledgementMode());
    }
    return topicSession;
  }
  public Topic getTopic() throws Exception {
    if (topic == null) {
      Object obj = getContext().lookup(topicName);
      topic = (Topic) obj;
    }
    return topic;
  }
  public TopicPublisher getTopicPublisher() throws Exception {
    if (topicPublisher == null) {
      topicPublisher = getTopicSession(false).createPublisher(getTopic());
    }
    return topicPublisher;
  }
  public TopicSubscriber getTopicSubscriber() throws Exception {
    if (topicSubscriber == null) {
      if (isDurable()) {
        topicSubscriber = getTopicSession(true).createDurableSubscriber(getTopic(), getDurableName());
      }
      else {
        topicSubscriber = getTopicSession(true).createSubscriber(getTopic());
      }
      topicSubscriber.setMessageListener(this);
      getTopicConnection(true).start();
    }
    return topicSubscriber;
  }
  public void publishText(String message) throws Exception {
    TextMessage textMessage = getTopicSession(false).createTextMessage();
    textMessage.clearBody();
    textMessage.setText(message);
    getTopicPublisher().publish(textMessage);
    if (isTransacted()) {
      getTopicSession(false).commit();
    }
  }
  public void onMessage(Message message) {
    if (message instanceof BytesMessage) {
      BytesMessage bytesMessage = (BytesMessage) message;
      //Process bytesMessage here
    }
    else {
      if (message instanceof MapMessage) {
        MapMessage mapMessage = (MapMessage) message;
        //Process mapMessage here
      }
      else {
        if (message instanceof ObjectMessage) {
          ObjectMessage objectMessage = (ObjectMessage) message;
          //Process objectMessage here
        }
        else {
          if (message instanceof StreamMessage) {
            StreamMessage streamMessage = (StreamMessage) message;
            //Process streamMessage here
          }
          else {
            if (message instanceof TextMessage) {
              TextMessage textMessage = (TextMessage) message;
              try{
                System.out.println("Message content is:" + textMessage.getText());
                } catch(JMSException e){
                     e.printStackTrace();
                  }

            }
          }
        }
      }
    }
    try {
      if (isTransacted()) {
        getTopicSession(false).commit();
      }
    }
    catch(Exception ex) {
      ex.printStackTrace();
    }
  }
  public void close() throws Exception {
    if (topicPublisher != null) {
      topicPublisher.close();
    }
    if (topicSubscriber != null) {
      topicSubscriber.close();
    }
    if (topicSession != null) {
      topicSession.close();
    }
    if (topicConnection != null) {
      topicConnection.close();
    }
  }
}