fetchmessagetask.java.svn-base

监控整个SP系统的SMS信息通道的发送情况并且生成日志。 支持权限判断。

package com.infobank.superchannel.daemon.messagelauncher;

import java.rmi.RemoteException;
import java.util.LinkedList;
import java.util.Queue;

import com.infobank.superchannel.daemon.databasecache.DatabaseCache;
import com.infobank.superchannel.pojo.TextMessage;
import com.infobank.superchannel.util.DaemonConfiguration;

/**
 * @author Xiaoguang Sun
 * @version 类说明: 从数据库缓冲服务器获取新待发信息任务
 */
public class FetchMessageTask implements Runnable {
  // 消息发送服务器对象
  private MessageLauncherDaemon daemon=null;
  // 每次重新读取新信息的间隔时间
  private int interval=0;
    // 消息标签
    private String tag=null;

  /**
   * 构造函数
   * @param daemon 消息发送服务器对象
   */
  public FetchMessageTask(MessageLauncherDaemon daemon){
    this.daemon=daemon;
        this.tag=daemon.getTag();

        StringBuilder sb=new StringBuilder();
        sb.append("/daemon[@type='MessageLauncher' and @tag='");
        sb.append(tag);
        sb.append("']/task/fetcher/interval");
    interval=DaemonConfiguration.getInstance().getInt(sb.toString());
  }

  /**
   * 待发消息读取任务执行体
   */
  public void run(){
    String lastMessageContent="";
    while(daemon.isRunnable()){
      try{
        // 从数据库缓冲获取新待发消息
        Queue<TextMessage> messages=daemon.fetchNewMessage();
        if(messages==null)
          continue;
        
        while(daemon.isMessageQueueFull()){
          Thread.sleep(interval);
        }
        // 将信息打包成为信息桶，同一个信息桶内存放的消息内容是完全相同的，用于判断信息是否为批量
        Queue<TextMessage> bucket=new LinkedList<TextMessage>();
        lastMessageContent="";
        for(TextMessage message:messages){
          if(message.getMessage().equals(lastMessageContent)){
            bucket.add(message);
          }else{
            lastMessageContent=message.getMessage();
            if(bucket.size()>0){
              // this is not the first time when lastMessageContent=""
              daemon.addMessage(bucket);
              bucket=new LinkedList<TextMessage>();
              bucket.add(message);
            }else{
              // this is the first time when lastMessageContent=""
              bucket.add(message);
            }
          }
        }
        if(bucket.size()>0)
          daemon.addMessage(bucket);
        Thread.sleep(interval);
      }catch(InterruptedException ex){
        ex.printStackTrace();
      }
    }
  }
}