slipwindowctrl.java

来自「短信系统SMS：支持普通短信、长短信和wap push短信的发送。」· Java 代码 · 共 128 行

package com.ekun.common.sms;

import java.util.*;

import com.ekun.common.log.*;

public class SlipWindowCtrl
{
  private static Logger logger = LoggerFactory.getLogger(SlipWindowCtrl.class);
  //滑动窗口大小，即：以发送但没有收到应答的包个数。
  private int slipWindowSize = 16;
  //等待响应的超时，单位：毫秒，一般情况下，对于超时的需要重新发送！
  private int waitACKTimeout = 60 * 1000;
  private HashMap hsPending;
  public SlipWindowCtrl()
  {
    hsPending = new HashMap(100);
  }
  /**
   * 加入未决队列,等待ACK返回
   * @param key String
   * @param objReq SmsInfo：要加入滑动窗口的request对象！
   * @param isBlock int[]: 滑动窗口是否阻塞
   * @return SmsInfo： 正常情况下返回null,若response比较快是，可能在req写入到滑动队列前已经等待，此时，返回等待的对象！
   */
  public synchronized SmsInfo pushCMDPending(String key, SmsInfo objReq)
  {
    try
    {
      //记录进入时间
      objReq.liveTime = System.currentTimeMillis();
      //step 1:察看队列中是否已经有返回的数据!
      SmsInfo myobj = (SmsInfo) hsPending.remove(key);
      //如果没有,则证明respose还没有返回!(正常情况下应该这样!)
      if(myobj == null)
      {
        //加入未决队列
        this.hsPending.put(key, objReq);
        //若超过滑动窗口大小限制,则等待ACK返回.
        if (hsPending.size() > this.slipWindowSize)
        {
          wait(waitACKTimeout);
        }
      }
      else
      {
        //如果此时response已经返回，则直接notify并返回。
        return myobj;
      }
    }
    catch (InterruptedException ex)
    {
      logger.error("SlipWindowCtrld.pushCMDPending Error：" + ex.getMessage(),
                   ex);
    }
    return null;
  }

  /**
   * 当收到ACK后查找原来的请求,若查到通知滑动窗口继续接受,否则丢弃!
   * @param key String
   * @param smsInfo SmsInfo:此参数只为CMPP可用,用以在Response早与request进入的情况下，记录submit response回来后的MsgID。
   * @return SmsInfo
   */
  public synchronized SmsInfo popPending(String key, SmsInfo smsInfo)
  {
    //从未决队列中删除
    SmsInfo myobj = (SmsInfo) hsPending.remove(key);
    //若未决队列中没有匹配到,则可能是ACK先到,加入未决队列！
    if (null == myobj)
    {
      smsInfo.liveTime = System.currentTimeMillis();
      hsPending.put(key, smsInfo);
      logger.debug("滑动窗口没有匹配到返回的数据，可能response早于request到达! Key: " + key);
    }
    notifyAll();
    return myobj;
  }

  /**
   * 取出所有未决的对象! 当重新建立连接时或者阻塞超时时,取出并重新发!
   * @return Object[]
   */
  public synchronized ArrayList getPending()
  {
    Iterator it = hsPending.values().iterator();
    SmsInfo myobj = null;
    ArrayList alist = new ArrayList();
    boolean havePend = false;
    while (it.hasNext())
    {
      myobj = (SmsInfo) it.next();
      if(myobj == null)
      {
        it.remove();
        havePend = true;
        continue;
      }
      //判断是否超时!
      if ((System.currentTimeMillis() - myobj.liveTime) >= waitACKTimeout)
      {
        it.remove();
        havePend = true;
        alist.add(myobj);
      }
    }
    if(havePend)
      notifyAll();
    return alist;
  }

  public int getPendingSize()
  {
    return hsPending.size();
  }

  public int getSlipWindowSize()
  {
    return slipWindowSize;
  }

  public void setSlipWindowSize(int slipWindowSize)
  {
    this.slipWindowSize = slipWindowSize;
  }

}