jmsfront.java

采用jms进行通讯的源代码,采用weblogic及websphere方式进行通讯

/*
 * jmsFront.java
 *
 * Created on 2003年7月19日, 上午11:26
 */

package collector.jms;

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

import collector.common.*;
import collector.common.queueobject.*;
import pbs.service.struct.*;

/**
 *
 * @author  WangJun
 */
public class JmsFront
    implements MessageListener {
  private Context m_InitialContext = null;
  private TopicConnectionFactory m_TopicConnectionFactory = null;
  private Topic m_Topic = null;
  private TopicSession m_TopicSession = null;
  private TopicConnection m_TopicConnection = null;
  private TopicPublisher m_TopicPublisher = null;
  private TopicSubscriber m_TopicSubscriber = null;

  private ArrayList m_NotifyQueue = null;
  private ArrayList m_CollectorQueue = null;

  private String m_LocalAddress = "";

  private int m_SendTimes = 0;
  private int m_RecTimes = 0;

  private String m_AppserverType = " ";
  /** Creates a new instance of jmsFront */
  public JmsFront(String m_LocalAddrsss) {
    this.m_NotifyQueue = new ArrayList();
    this.m_CollectorQueue = new ArrayList();

    this.m_LocalAddress = getLocalHostName();
  }

  private String getLocalHostName() {
    String m_TempLocalName = "";
    try {
      InetAddress ia = InetAddress.getLocalHost();
      m_TempLocalName = ia.getHostName();
    }
    catch (Exception e) {
      CollectorDefine.SystemPrintln("getLocalHostName In JmsComm Error #1" +
                                    e.toString());
      return "##";
    }

    return m_TempLocalName;
  }

  public int getNotifyQueueNum() {
    if (this.m_NotifyQueue.isEmpty() == true) {
      return -1;
    }

    return this.m_NotifyQueue.size();
  }

  private int putNotifyQueue(Object m_Object) {
    this.m_NotifyQueue.add(m_Object);
    return 1;
  }

  public Object getNotifyQueue() {
    // wj modi at hangzhou 20040602
    //return (Object)this.m_NotifyQueue.remove(0);
    if (this.m_NotifyQueue.size() > 0) {
      return (Object)this.m_NotifyQueue.remove(0);
    }
    return null;
    // wj modi at hangzhou 20040602
  }

  public int removeAllNotifyQueue() {
//    this.m_NotifyQueue.clear();
    if (CollectorDefine.APPSERVER_CLUSTER == 1) {
//    if (m_AppserverType.toLowerCase().equals("ibm") != true) {
      this.m_NotifyQueue.clear();
    }

    return 1;
  }

  public int getCollectorQueueNum() {
    if (this.m_CollectorQueue.isEmpty() == true) {
      return -1;
    }

    return this.m_CollectorQueue.size();
  }

  private int putCollectorQueue(Object m_Object) {
    this.m_CollectorQueue.add(m_Object);
    return 1;
  }

  public Object getCollectorQueue() {

    // wj modi at hangzhou 20040602
    //return (Object)this.m_CollectorQueue.remove(0);
    if (this.m_CollectorQueue.size() > 0) {
      return (Object)this.m_CollectorQueue.remove(0);
    }
    return null;
    // wj modi at hangzhou 20040602

  }

  public int removeAllCollectorQueue() {
//    this.m_CollectorQueue.clear();
    if (CollectorDefine.APPSERVER_CLUSTER == 1) {
      this.m_CollectorQueue.clear();
    }

    return 1;
  }

  public int initJMS(String m_AppserverType, String m_InitContextFactory,
                     String m_JmsURL, String m_JmsPort,
                     String m_TopicFactoryName, String m_TopicName) {
    this.m_AppserverType = m_AppserverType;
    if (m_AppserverType.toLowerCase().equals("ibm") == true) {
      try {
        Properties namingProps = new Properties();
        namingProps.put(javax.naming.Context.PROVIDER_URL, m_JmsURL + m_JmsPort);
        namingProps.put(javax.naming.Context.INITIAL_CONTEXT_FACTORY,
                        m_InitContextFactory);
        m_InitialContext = new InitialContext(namingProps);

        m_TopicConnectionFactory = (TopicConnectionFactory) m_InitialContext.
            lookup(m_TopicFactoryName);
        m_Topic = (Topic) m_InitialContext.lookup(m_TopicName);
      }
      catch (NamingException e) {
        return -1;
      }
    }
    else
    if (m_AppserverType.toLowerCase().equals("weblogic") == true) {
      try {
        //System.out.println(
        //  "in jmsfront, try to build one m_TopicConnection of jms");
        Properties namingProps = new Properties();

        namingProps.put(javax.naming.Context.INITIAL_CONTEXT_FACTORY,
                        m_InitContextFactory);
        namingProps.put(javax.naming.Context.PROVIDER_URL,
                        "t3://" + m_JmsURL + ":" + m_JmsPort);

        /*namingProps.put(javax.naming.Context.PROVIDER_URL,
                        "iiop://" + m_JmsURL + ":" + m_JmsPort + "/");
         */

        m_InitialContext = new InitialContext(namingProps);
        m_Topic = (Topic) m_InitialContext.lookup(m_TopicName);
        m_TopicConnectionFactory = (TopicConnectionFactory) m_InitialContext.
            lookup(m_TopicFactoryName);
      }
      catch (NamingException e) {
        //System.out.println("initJMS in JmsFront #1"+e.toString ());
        return -1;
      }
    }
    else {
      try {
        Properties namingProps = new Properties();
        //env.put("java.naming.factory.m_InitialContext","com.sun.jndi.cosnaming.CNCtxFactory");
        //env.put("java.naming.provider.url", "iiop://badger.prc.sun.com:3700");
        namingProps.setProperty("java.naming.factory.m_InitialContext",
                                m_InitContextFactory);
        namingProps.setProperty("org.omg.CORBA.ORBInitialHost", m_JmsURL);
        namingProps.setProperty("org.omg.CORBA.ORBInitialPort", m_JmsPort);
        m_InitialContext = new InitialContext(namingProps);
        m_Topic = (Topic) m_InitialContext.lookup(m_TopicName);
        m_TopicConnectionFactory = (TopicConnectionFactory) m_InitialContext.
            lookup(m_TopicFactoryName);
      }
      catch (NamingException e) {
        return -1;
      }
    }

    try {
      m_TopicConnection = m_TopicConnectionFactory.createTopicConnection();
      m_TopicSession = m_TopicConnection.createTopicSession(false,
          Session.AUTO_ACKNOWLEDGE);
      m_TopicPublisher = m_TopicSession.createPublisher(m_Topic);
      m_TopicSubscriber = m_TopicSession.createSubscriber(m_Topic);
      m_TopicSubscriber.setMessageListener(this);
      m_TopicConnection.start();
    }
    catch (JMSException e) {
      return -1;
    }

    CollectorDefine.SystemPrintln("Jms Init Success ! m_JmsURL:" +
                                  m_JmsURL + "m_JmsPort :" + m_JmsPort +
                                  "m_TopicFactoryName:" +
                                  m_TopicFactoryName + "m_TopicName:" +
                                  m_TopicName);

    return 1;
  }

  public int reConnect() {

    try {
      if (m_TopicConnectionFactory == null) {
      }

      m_TopicConnection = m_TopicConnectionFactory.createTopicConnection();
      m_TopicSession = m_TopicConnection.createTopicSession(false,
          Session.AUTO_ACKNOWLEDGE);
//         this.m_TopicPublisher = this.m_TopicSession.createPublisher (this.m_TargetT);
//          m_TopicSubscriber = m_TopicSession.createSubscriber (m_Topic);
//          m_TopicSubscriber.setMessageListener (this);
      m_TopicConnection.start();
    }
    catch (JMSException e) {
      return -1;
    }

    return 1;
  }

  public int broadcast(JmsObject m_pJmsObject) {
    int ret;
    ObjectMessage m_ObjectMessage = null;

    //System.out.println(" begin send msg");
    if (this.m_TopicSession != null) {
      try {
        this.m_TopicPublisher = this.m_TopicSession.createPublisher(this.
            m_Topic);
        m_ObjectMessage = this.m_TopicSession.createObjectMessage();

        m_ObjectMessage.setObject(m_pJmsObject);
        this.m_TopicPublisher.publish(m_ObjectMessage);
        this.m_TopicPublisher.close();
        this.m_TopicPublisher = null;

        this.m_SendTimes++;
        if (this.m_SendTimes > JmsDefine.MAX_SEND_TIMES) {
          this.m_SendTimes = 0;
        }
      }
      catch (JMSException m_JMSException) {
        CollectorDefine.SystemPrintln("Jms Exception :" +
                                      m_JMSException.toString());
        return JmsDefine.JMS_EXCEPTION;
      }
    }
    //System.out.println(" end send msg");
    return 1;
  }

  public void onMessage(javax.jms.Message m_ObjectMessage) {
    ObjectMessage msg = null;
    Object m_Object = null;
    ParamModify m_ParamModify = null;
    int i;

    try {
      if (m_ObjectMessage instanceof ObjectMessage) {
        msg = (ObjectMessage) m_ObjectMessage;
        m_Object = (Object) msg.getObject();
        if (m_Object == null) {
          return;
        }

        if (m_Object instanceof ParamModify) {
          m_ParamModify = (ParamModify) msg.getObject();
          CollectorDefine.SystemPrintln("table_no=====" +
                                        (new Long(m_ParamModify.m_table_no)).
                                        toString());
          for (i = 0; i < m_ParamModify.m_ids.length; i++) {
            CollectorDefine.SystemPrintln("id=====" +
                                          (new Long(m_ParamModify.m_ids[i])).
                                          toString());
          }

          this.m_RecTimes++;
          if (this.m_RecTimes > JmsDefine.MAX_SEND_TIMES) {
            this.m_RecTimes = 0;
          }

          putNotifyQueue(m_ParamModify);
        }
        else if (m_Object instanceof JmsObject) {
          JmsObject m_JmsObject = (JmsObject) m_Object;
          String m_TempHostName = m_JmsObject.getHostName();
          if (m_TempHostName.toLowerCase().equals(this.m_LocalAddress.
                                                  toLowerCase()) != true) {
            putCollectorQueue(m_JmsObject);

            this.m_RecTimes++;
            if (this.m_RecTimes > JmsDefine.MAX_SEND_TIMES) {
              this.m_RecTimes = 0;
            }
          }
        }
      }
    }
    catch (Exception e) {
//        e.printStackTrace();
    }

    return;
  }

  /** Getter for property m_RecTimes.
   * @return Value of property m_RecTimes.
   */
  public int getRecTimes() {
    return m_RecTimes;
  }

  /** Setter for property m_RecTimes.
   * @param m_RecTimes New value of property m_RecTimes.
   */
  public void setRecTimes(int m_RecTimes) {
    this.m_RecTimes = m_RecTimes;
  }

  /** Getter for property m_SendTimes.
   * @return Value of property m_SendTimes.
   */
  public int getSendTimes() {
    return m_SendTimes;
  }

  /** Setter for property m_SendTimes.
   * @param m_SendTimes New value of property m_SendTimes.
   */
  public void setSendTimes(int m_SendTimes) {
    this.m_SendTimes = m_SendTimes;
  }

  public int close() {
    try {

      if (m_TopicSubscriber != null) {
        m_TopicSubscriber.close();
        m_TopicSubscriber = null;
      }

      if (m_TopicPublisher != null) {
        m_TopicPublisher.close();
        m_TopicPublisher = null;
      }

      if (m_TopicSession != null) {
        m_TopicSession.close();
        m_TopicSession = null;
      }

      if (m_TopicConnection != null) {
        m_TopicConnection.close();
        m_TopicConnection = null;
      }
    }
    catch (JMSException m_JMSException) {
      CollectorDefine.SystemPrintln(m_JMSException.getErrorCode());
    }

    return 1;
  }

}