spservice.java

采用JAVA开发

package com.gctech.cmpp.server;

import java.net.*;
import java.io.*;

import org.apache.log4j.Logger;

import com.gctech.cmpp.server.dao.SpInfoDao;
import com.gctech.cmpp.server.vo.SpInfoVo;
import com.gctech.util.net.SocketService;
import com.gctech.util.Tools;
import com.gctech.util.MD5;
import com.gctech.cmpp.util.*;
import com.gctech.cmpp.msg.*;


/**
 * <p>Title: 短信 API</p>
 * <p>Description: 短信SP API</p>
 * <p>Copyright: Copyright (c) 2004</p>
 * <p>Company: GCTECH</p>
 * @author 王红宝
 * @version $Id: SpService.java,v 1.1 2004/05/21 07:50:11 wanghb Exp $
 */

public class SpService extends SocketService {
  public static long ACTIVE_TEST_INTERVAL = 18000;
  public static int TIME_OUT = 18000;
  void close(){
    this.running = false;
  }
  void activeTest(){
    logger.debug("active test for "+this.sourceAddr);
    if ( (System.currentTimeMillis() - latestComTime) >
        ACTIVE_TEST_INTERVAL){
      ActiveTestRequest active = new ActiveTestRequest();
      active.getHead().setSequenceId(IDGenerator.getInstance().increment());
      MsgLogger.log("发送消息："+active);
      send(active.toByteArray());
    }
  }

  //最后一层通讯时间，用来控制链路检测包
  long latestComTime;
  //源地址，spid,例如6299
  String sourceAddr;
  //输出流
  OutputStream out;
  //输入流
  InputStream in;
  //客户端IP
  String ip;
  //关闭标签
  boolean running;
  //服务器ID,暂时没用
  int serviceId;

  public SpService() {
  }

  void authen(ConnectRequest conReq, ConnectResponse conResp){
    conResp.setSequence(conReq);
    this.sourceAddr = conReq.getSourceAddr();
    //查询出SP信息。
    SpInfoVo spInfo = SpInfoDao.findByPk(sourceAddr);
    if ( spInfo == null ){
      logger.warn(sourceAddr+" not found!");
      conResp.setStatus(Status.CONNECT_ILLEGAL_SOURCE);
    }else if ( ip.equals(spInfo.getBindAddr())){
      //取出时间戳
      String timestamp = "" + conReq.getTimestamp();
      if ( timestamp.length() < 10 ){
        timestamp = "0" + timestamp;
      }
      //计算明文长度
      byte[] sharedSecret = spInfo.getSharedSecret().getBytes();
      byte[] spId = spInfo.getSpId().getBytes();
      byte[] md5Msg = Util.md5Connect(spId, sharedSecret, timestamp.getBytes());
      //校验md5
      boolean authen = true;
      for ( int i = 0; i < 16; i++ ){
        if ( md5Msg[i] != conReq.getAuthenSource()[i] ){
          authen = false;
          break;
        }
      }
      //如果认证通过，生成服务端认证码
      if (authen){
        byte[] authSrc = conReq.getAuthenSource();
        byte[] authenIsmg = Util.md5ConnectResponse(Status.SUCCESS,
            authSrc, sharedSecret);
        conResp.setAuthenISMG(authenIsmg);
      } else{
        conResp.setStatus(Status.CONNECT_AUTHEN_ERROR);
      }
    } else{
      logger.warn(sourceAddr + " login from "+ip+",but the correct ip is"
                  + spInfo.getBindAddr());
      conResp.setStatus(Status.CONNECT_ILLEGAL_SOURCE);
    }
  }
  synchronized void send(byte[] data) {
    try {
      out.write(data);
    }
    catch (IOException ex) {
      logger.error(ex, ex);
      running = false;
    }
    this.latestComTime = System.currentTimeMillis();
  }

  public void run() {
    ip = "UNKNOWN";
    sourceAddr = "UNKNOWN";
    try {
      ip = sock.getInetAddress().getHostAddress();
      sock.setSoTimeout(TIME_OUT);
      sock.setKeepAlive(true);
      in = sock.getInputStream();
      ConnectRequest conReq = new ConnectRequest();
      byte[] data = new byte[conReq.getHead().getTotalLength()];

      int rt = in.read(data);
      ConnectResponse conResp = new ConnectResponse();

      if ( rt != conReq.getHead().getTotalLength() ){
        logger.warn("从" + ip + "的连接包数据长度错误:" + rt);
        conResp.setStatus(Status.CONNECT_WRONG_STRUCT);
      } else{
        int cmdId = Tools.byte2int(data, 4);
        if ( cmdId == CommandID.CMPP_CONNECT ){
          conReq.fromByteArray(data);
          MsgLogger.log(conReq);
          authen(conReq, conResp);
        }else{
          logger.warn("连接CommandId错误:"+cmdId);
          conResp.setStatus(Status.CONNECT_WRONG_STRUCT);
        }
      }
      out = sock.getOutputStream();

      MsgLogger.log(conResp);
      send(conResp.toByteArray());
      //如果认证通过
      if (conResp.getStatus() == Status.SUCCESS ){
        //注册sp，开始服务
        serviceId = IDGenerator.getInstance().increment();
        Object old = Server.getInstance().register(this);
        //如果有旧的连接，把它关闭。
        if (old != null){
          SpService oldSvc = (SpService)old;
          oldSvc.close();
        }
        doService();
      }
    } catch (SocketException ex) {
      logger.error(ex, ex);
    } catch(IOException ioe){
      logger.error(ioe, ioe);
    } catch(InterruptedException ie){
      logger.error(ie, ie);
    } catch(Exception ee){
      logger.error(ee, ee);
    } finally{
      if ( sock != null ){
        try {
          sock.close();
        }
        catch (IOException ex1) {
          logger.error(ex1, ex1);
        }
      }
      //注销服务
      Server.getInstance().unregister(this);
      logger.info("释放从[ip:"+ip+",spId:"+sourceAddr+"]过来的连接。");
    }
  }


  //服务函数
  /**
  void doService() throws InterruptedException, Exception {
    running = true;
    CmppMsg cmppmsg=new CmppMsg();
    CmppMsg tmp=null;

    SubmitRequest submitrequest=null;
    DeliverResponse deliverresponse=null;
    ActiveTestRequest activerequest=null;
    ActiveTestResponse activeresponse=null;

    while ( sock.isConnected() && running ){
      try {
        /**delete by liya begin
        //读数据
        byte[] bLen = new byte[4];
        int rt = in.read(bLen);
        if (rt != 4) {
          throw new IOException("读取包头长度错误:" + rt);
        }
        int length = Tools.byte2int(bLen);
        if (length < 12 || length > 2048) {
          throw new IOException("读取数据长度错误:" + length);
        }
        length -= 4;
        //读取有效数据长度。
        byte[] data = new byte[length];
        rt = in.read(data);
        if (rt != length) {
          throw new IOException("读取数据长度不符合:" + rt + " vs " + length);
        }
        this.latestComTime = System.currentTimeMillis();
        int cmdId = Tools.byte2int(data);
        delete by liya end
        tmp=cmppmsg.read(in);
        switch (tmp.getCommandId()) {
          case CommandID.CMPP_SUBMIT:
            submitrequest=(SubmitRequest)tmp;
            MsgLogger.log(submitrequest);
            SubmitResponse sr=new SubmitResponse(submitrequest.getHead());
            sr.write(out);
            //发送到6299的接口
            SmsInterface.getInstance().submit(submitrequest);
            break;
          //链路检测
          case CommandID.CMPP_ACTIVE_TEST:
            activerequest=(ActiveTestRequest)tmp;
            ActiveTestResponse atr=new ActiveTestResponse(activerequest.getHead());
            MsgLogger.log(activerequest);
            atr.write(out);
            break;
          //链路检测回复
          case CommandID.CMPP_ACTIVE_TEST_RESP:
            activeresponse = (ActiveTestResponse)tmp;
            activeresponse.readbody();
            MsgLogger.log(activeresponse);
            break;
          //上行回复
          case CommandID.CMPP_DELIVER_RESP:
            deliverresponse=(DeliverResponse)tmp;
            deliverresponse.readbody();
            MsgLogger.log(deliverresponse);
            break;
          default:
            throw new IOException("CommandID Error:" + tmp.getCommandId());
        }
        //MsgLogger.logMsg(length, data, 0, this.sourceAddr);
      }
      catch (SocketTimeoutException ex) {
        logger.debug("waiting data time out!");
      }
    }
  }
  */

  void doService() throws InterruptedException, IOException {
     running = true;
     while ( sock.isConnected() && running ){
       try {
         //读数据
         byte[] bLen = new byte[4];
         int rt = in.read(bLen);
         if (rt != 4) {
           throw new IOException("读取包头长度错误:" + rt);
         }
         int length = Tools.byte2int(bLen);
         if (length < 12 || length > 2048) {
           throw new IOException("读取数据长度错误:" + length);
         }
         length -= 4;
         //读取有效数据长度。
         byte[] data = new byte[length];
         rt = in.read(data);
         if (rt != length) {
           throw new IOException("读取数据长度不符合:" + rt + " vs " + length);
         }
         this.latestComTime = System.currentTimeMillis();
         int cmdId = Tools.byte2int(data);
         switch (cmdId) {
           case CommandID.CMPP_SUBMIT:
             //记录日志
             SubmitRequest subReq = new SubmitRequest();
             subReq.getHead().setTotalLength(data.length+4);
             subReq.fromByteArray(data, 0);
             MsgLogger.log(subReq);
             //回复SubmitResponse @todo 如何产生msgId，并和subReq对应起来。
//             long id=HomeWayMO.idgenerator.nextId();
//             byte[] bid=HomeWayMO.idgenerator.LongToBytes8(id);
             byte[] bid = null;
             SubmitResponse subResp =new SubmitResponse(bid);
             subResp.setSequence(subReq);
             subResp.getHead().setTotalLength(21);
             subResp.setStatus((byte)0x00);
             SubmitResp(subResp);
             //发送消息
//             SmsInterface.getInstance().submit(subReq,bid);
             break;
           //链路检测
           case CommandID.CMPP_ACTIVE_TEST:
             ActiveTestRequest activeReq = new ActiveTestRequest();
             activeReq.fromByteArray(data, 0);
             MsgLogger.log("接收消息：" + activeReq);
             activeTestResp(activeReq);
             break;
           //链路检测回复
           case CommandID.CMPP_ACTIVE_TEST_RESP:
             ActiveTestResponse activeResp = new ActiveTestResponse();
             activeResp.fromByteArray(data, 0);
             MsgLogger.log("接收消息："+activeResp);
             break;
           //上行回复。
           case CommandID.CMPP_DELIVER_RESP:
             DeliverResponse delResp = new DeliverResponse();
             delResp.fromByteArray(data, 0);
             MsgLogger.log(delResp);
             break;
           default:
             logger.warn("Receive UNKNOW CMD:"+cmdId);
             throw new IOException("包命令错误:" + cmdId);
         }
         //MsgLogger.logMsg(length, data, 0, this.sourceAddr);
       }
       catch (SocketTimeoutException ex) {
         logger.debug("waiting data time out!");
       }
     }
  }

  void activeTestResp(ActiveTestRequest request) throws IOException {
    ActiveTestResponse resp = new ActiveTestResponse();
    resp.setSequence(request);
    byte[] data = resp.toByteArray();
    MsgLogger.log("回复消息："+resp);
    send(data);
  }

  void SubmitResp(SubmitResponse response) throws IOException {
    byte[] data = response.toByteArray();
    MsgLogger.log("回复消息："+response);
    send(data);
  }

  static final Logger logger = Logger.getLogger(SpService.class);


}