phsclient.java

JAVA实现的短信网关程序

package sms.PHS;

/**
 * <p>Title: </p>
 * <p>Description: </p>
 * <p>Copyright: Copyright (c) 2004</p>
 * <p>Company: </p>
 * @author not attributable
 * @version 1.0
 */
import java.io.IOException;
import java.io.PrintStream;
import java.net.Socket;
import java.util.*;
import java.sql.*;
import dbconnectionbean.*;

public class PHSClient
    extends Thread {
  private Thread activeThread;
  private int activeTime;
  private boolean debug;
  private LinkedList listeners;
  private Hashtable lockMap;
  protected PHSIO m_io;
  private int status;
  private Hashtable waitPackets;

  Connection con_cp = null;
  ConnectionPoolBean beanCP;
  class MOThread
      extends Thread {
    public void initCon() {
       beanCP = new ConnectionPoolBean();
       beanCP.setinit();
       beanCP.setName("smgpsp");
       con_cp = beanCP.getSqlCon();
       System.out.println("(^_^)连接初始化：：初始化tyfo网关接口数据库连接完成。" + "\n");

     }
     public String bytesToBCD(byte[] b) {
       StringBuffer sb = new StringBuffer();
       for (int i = 0; i < b.length; i++) {
         sb.append( (b[i] >> 4) & 0x0f);
         sb.append(b[i] & 0x0f);
       }
       return sb.toString();
     }

    public void run() {
      initCon();
      boolean loop = true;
      try {
        do {


          PHSMessage msg = new PHSMessage(m_io);
          if (msg == null) {
            log("the MO do not get a msg", debug);
            return;
          }
          log("get one mo msg:" + msg.getCommandId(), debug);
          if ( (msg.getCommandId() & 0x80000000) != 0) {
            if (msg != null &&
                lockMap.containsKey(new Integer(msg.getSequenceId()))) {
              addWaitPacket(msg);
              Object lock = lockMap.remove(new Integer(msg.getSequenceId()));
              synchronized (lock) {
                System.out.println("------------1---------------");
                lock.notify();
              }
              continue;
            }

          }
          else {
            //System.out.println("test:" + msg.getCommandId());
            switch (msg.getCommandId()) {
              case 6: { // '\006'   exit
                PHSExit exit = new PHSExit(msg);
                log("==============receive exit message:============", debug);
                PHSExitRep resp = new PHSExitRep();
                resp.setSequenceId(exit.getSequenceId());
                putRequest(resp);
                loop = false;
                break;
              }

              case 4: { // '\004'     active_test
                PHSActiveTest test = new PHSActiveTest(msg);
                log("================receive active message:========", debug);
                PHSActiveTestRep resp = new PHSActiveTestRep();
                resp.setSequenceId(test.getSequenceId());
                putRequest(resp);
                break;
              }

              case 3: { // '\003'  deliver_resp
                PHSDeliver deliver = new PHSDeliver(msg);


                //log("receive deliver message:", debug);
                //log("getDestTermId=" + deliver.getDestTermId(), debug);
                //log("getSrcTermId=" + deliver.getSrcTermId(), debug);
                //log("getIsReport=" + deliver.getIsReport(), debug);
                //log("getMsgFormat=" + deliver.getMsgFormat(), debug);
                //log("getMsgLength=" + deliver.getMsgLength(), debug);
                //log("getRecvTime=" + deliver.getRecvTime(), debug);
                //log("getMsgId=" + deliver.getMsgId(), debug);
                if (deliver.getMsgFormat() == 15) {
                  //log("getMsgContent=" +
                  //    new String(deliver.getMsgContent(), "GBK"),
                  //    debug);
                }
                else {
                  //log("getMsgContent=" +
                  //    new String(deliver.getMsgContent(), "GBK"),
                  //    debug);

                }
                byte Result = 0;
                /*
                     if (!deliver.getDestTermId().trim().equals("1189016")) {
                  System.out.println(deliver.getDestTermId().trim().equals(
                      "1189016"));
                  Result = 0x27;
                                 }
                 */

                String sMsgID=bytesToBCD(deliver.getMsgId().getBytes());
                int nIsReport=deliver.getIsReport();
                int nMsgFormat=deliver.getMsgFormat();
                //int sequenceid=deliver.getSequenceId();
                String sRecvTime=deliver.getSrcTermId().trim();
                String sSrcTermID=deliver.getSrcTermId().trim();
                String sDestTermID=deliver.getDestTermId().trim();
                int nMsgLength=deliver.getMsgLength();
                String sMsgContent=new String(deliver.getMsgContent());

                String sql="";

                //sql="insert into phs_deliver (sMsgID,nIsReport,nMsgFormat,sRecvTime,sSrcTermID,sDestTermID,nMsgLength,sMsgContent,sequenceid) values ('"+sMsgID+"','"+nIsReport+"','"+nMsgFormat+"','"+sRecvTime+"','"+sSrcTermID+"','"+sDestTermID+"','"+nMsgLength+"','"+sMsgContent+"','"+sequenceid+"')";
                System.out.println("得到MO数据信息..........");



                sql="insert into phs_deliver (sMsgID,nIsReport,nMsgFormat,sRecvTime,sSrcrmID,sDestTermID,nMsgLen,sMsgContent) values ('"+sMsgID+"','"+nIsReport+"','"+nMsgFormat+"','"+sRecvTime+"','"+sSrcTermID+"','"+sDestTermID+"','"+nMsgLength+"','"+sMsgContent+"')";
                System.out.print("写mo数据的sql:"+sql);
                try
                  {
                    Statement smt=con_cp.createStatement();
                    if(nIsReport==0)
                    {
                      //smt.executeUpdate(sql);
                      System.out.print("得到上行数据:"+sql);
                    }
                    else
                    {
                    //是状态报告，写入phs_report表里
                    String reportstr=sMsgContent;
                    String report_id="";
                    String sub="";
                    String dlvrd="";
                    String submit_date="";
                    String done_date="";
                    String stat="";
                    String err="";
                    String txt="";
                    String ssDestTermID=sDestTermID;
                    String msgcont="no content";

                    report_id=reportstr.substring(reportstr.indexOf("id:"),reportstr.indexOf("sub:"));
                    report_id=bytesToBCD(report_id.getBytes());
                    sub=reportstr.substring(reportstr.indexOf("sub:")+4,reportstr.indexOf("dlvrd:"));
                    dlvrd=reportstr.substring(reportstr.indexOf("dlvrd:")+6,reportstr.indexOf("submit date:"));
                    submit_date=reportstr.substring(reportstr.indexOf("submit date:")+13,reportstr.indexOf("done date:"));
                    done_date=reportstr.substring(reportstr.indexOf("done date:")+10,reportstr.indexOf("stat:"));
                    stat=reportstr.substring(reportstr.indexOf("stat:")+5,reportstr.indexOf("err:"));
                    err=reportstr.substring(reportstr.indexOf("err:")+4,reportstr.indexOf("Text:"));
                    txt="no report content";

                    sql="insert into phs_report (report_id,sub,dlvrd,submit_date,done_date,stat,err,txt,sDestTermID,msgcont) values ('"+report_id+"','"+sub+"','"+dlvrd+"','"+submit_date+"','"+done_date+"','"+stat+"','"+err+"','"+txt+"','"+ssDestTermID+"','"+msgcont+"')";
                    System.out.print("sql:"+sql);
                    smt.executeUpdate(sql);
                    beanCP.setfreeconnection();
                    }
                  }catch(Exception e)
                  {
                  e.printStackTrace();
                  }

                PHSDeliverRep resp = new PHSDeliverRep(deliver.getMsgId(),
                    Result);
                resp.setSequenceId(deliver.getSequenceId());
                putRequest(resp);
                fireReceiveMessage(deliver);
                break;
              }

              case 5: // '\005'
              default: {
                log("unknow Message=" + msg.getCommandId(), debug);
                break;
              }
            }

          }

        }
        while (loop);

      }
      catch (Exception e) {
        log(e.toString(), debug);
        log("MOServer.run :disconnected !", debug);
        try {
          m_io.close();
          fireDisconnected();

        }
        catch (Exception ex) {
          e.printStackTrace();
        }
        return;
      }

    }

    public MOThread() {
      super();
    }
  }

  class ActiveTestThread
      implements Runnable {

    public void run() {
      while (true) {
        try {
          if (status == 1) {
            PHSActiveTest active = new PHSActiveTest();
            PHSActiveTestRep rep = null;
            try {
              rep = PHSClient.this.active(active, 5000L);
              if (rep == null) {
                m_io.close();
              }
            }
            catch (PHSException ex1) {
              m_io.close();
              ex1.printStackTrace();
            }
            Thread.sleep(activeTime);
            System.out.println("activeTime=" + activeTime);
          }
        }
        catch (InterruptedException ex) {
          return;
        }
        catch (Exception ex) {
          ex.printStackTrace();
        }
      }
    }

    public ActiveTestThread() {
      super();
    }
  }

  public PHSClient(String host, int port) throws PHSException, CSocksException,
      IOException {
    lockMap = new Hashtable();
    waitPackets = new Hashtable();
    listeners = new LinkedList();
    activeTime = 20000;
    status = 0;
    debug = false;
    PHSClientInit(host, port);
  }

  public void PHSClientInit(String host, int port) throws IOException,
      CSocksException, PHSException, IOException {
    log("PHSClient:PHSClientInit: Creating socket host=" + host + ",port=" +
        port, debug);
    Socket m_sokSocket = CSocks.Socket(host, port);
    m_sokSocket.setSendBufferSize(0x19000);
    log("PHSClient:PHSClientInit: Creating in/output stream.", debug);
    m_io = new PHSIO(m_sokSocket, this);
    MOThread mo = new MOThread();
    mo.start();
  }

  public PHSActiveTestRep active(PHSActiveTest active, long waitTime) throws
      PHSException {
    PHSMessage rep = putRequestForRep(active, waitTime);
    if (rep == null) {
      return null;
    }
    else {
      return new PHSActiveTestRep(rep);
    }
  }

  public boolean addListener(PHSClientListener listener) {
    return listeners.add(listener);
  }

  public void addWaitPacket(PHSMessage packet) {
    waitPackets.put(new Integer(packet.getSequenceId()), packet);
  }

  public void close() {
    try {
      log("Sending terminate message.", debug);
      if (m_io != null && !m_io.isDisconnected()) {
        exit(new PHSExit(), 2000L);
      }
    }
    catch (Exception e1) {
      log("Sending terminate failed.", debug);
    }
    try {
      activeThread.interrupt();
      lockMap.clear();
      waitPackets.clear();
      listeners.clear();
      m_io.close();
    }
    catch (Exception e) {
      //System.out.println("PHSClient.java=========" + e.getMessage());
    }

  }

  public PHSExitRep exit(PHSExit exit, long waitTime) throws PHSException {
    PHSMessage rep = putRequestForRep(exit, waitTime);
    if (rep == null) {
      return null;
    }
    else {
      try {
        m_io.close();
      }
      catch (Exception e) {}
      return new PHSExitRep(rep);
    }
  }

  protected void finalize() {
    close();
  }

  protected void fireDisconnected() {

    ListIterator iter = listeners.listIterator();
    while (iter.hasNext()) {
      PHSClientListener listener = (PHSClientListener) iter.next();
      listener.onClientDisconnected();
    }

  }

  private void fireReceiveMessage(PHSDeliver deliver) {
    ListIterator iter = listeners.listIterator();
    while (iter.hasNext()) {
      PHSClientListener listener = (PHSClientListener) iter.next();
      listener.onDeliverPacketReceive(deliver);
    }

  }

  public boolean isConnected() {
    return!m_io.isDisconnected();
  }

  private void log(String str, boolean debug) {
    if (debug) {
      System.out.println(str);
    }
  }

  public PHSLoginRep login(PHSLogin login, long waitTime) throws PHSException {
    PHSMessage rep = putRequestForRep(login, waitTime);
    if (rep == null) {
      return null;
    }
    PHSLoginRep ret = new PHSLoginRep(rep);
    if (ret.getStatus() == 0) {
      status = 1;
    }
    return ret;
  }

  private void putRequest(PHSMessage msg) throws PHSException {
    msg.encodePacket(m_io);
  }

  private PHSMessage putRequestForRep(PHSMessage msg, long waitTime) throws
      PHSException {
    Object lock;
    PHSMessage resp;
    if (msg == null || waitTime < 0L) {
      return null;
    }

    lock = new Object();
    lockMap.put(new Integer(msg.getSequenceId()), lock);
    putRequest(msg);
    resp = removeWaitPacket(new Integer(msg.getSequenceId()));

    if (resp != null) {
      lockMap.remove(new Integer(msg.getSequenceId()));
    }
    else {
      try {
        Object obj = lock;
        synchronized (lock) {
          lock.wait(waitTime); //???
          System.out.println("------------2---------------");
        }
        resp = removeWaitPacket(new Integer(msg.getSequenceId()));
        lockMap.remove(new Integer(msg.getSequenceId()));
      }
      catch (Exception e) {
        e.printStackTrace();
      }
    }

    return resp;
  }

  public boolean removeListener(PHSClientListener listener) {
    return listeners.remove(listener);
  }

  public PHSMessage removeWaitPacket(Integer seq) {
    return (PHSMessage) waitPackets.remove(seq);
  }

  public void startActiveTest(int periodTime) {
    if (activeThread != null) {
      activeThread.interrupt();
    }
    activeTime = periodTime;
    activeThread = new Thread(new ActiveTestThread());
    activeThread.start();
  }

  public PHSSubmitRep submit(PHSSubmit submit, long waitTime) throws
      PHSException {
    PHSMessage rep = putRequestForRep(submit, waitTime);
    if (rep == null) {

      return null;
    }
    else {
      return new PHSSubmitRep(rep);
    }
  }

}