compounddatabasecachestatusobserver.java.svn-base

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

package com.infobank.superchannel.observer.implementation;

import java.util.Hashtable;
import java.util.LinkedList;
import java.util.List;
import java.util.UUID;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicReference;

import com.infobank.superchannel.observer.DatabaseCacheStatusObserver;
import com.infobank.superchannel.pojo.BlackList;
import com.infobank.superchannel.pojo.BlackListLog;
import com.infobank.superchannel.pojo.ControllerUserToken;
import com.infobank.superchannel.pojo.GenericLog;
import com.infobank.superchannel.pojo.TextMessage;
import com.infobank.superchannel.pojo.TextMessageLog;

/**
 * @author Xiaoguang Sun
 * @version 类说明: 复合式数据库缓冲服务器状态观察者
 */
public class CompoundDatabaseCacheStatusObserver implements
    DatabaseCacheStatusObserver {
  private AtomicReference<BlackList> blacklist=new AtomicReference<BlackList>(new BlackList(new LinkedList<String>()));
  private AtomicInteger incomingMessageQueueSize=new AtomicInteger(0);
  private AtomicInteger messageQueueSize=new AtomicInteger(0);
  private AtomicInteger numberOfFailedMessages=new AtomicInteger(0);
  private AtomicInteger numberOfSuccessfulMessages=new AtomicInteger(0);
  private AtomicInteger statusQueueSize=new AtomicInteger(0);

  //所有绑定的观察者
  private List<DatabaseCacheStatusObserver> observers=new LinkedList<DatabaseCacheStatusObserver>();

  //监控界面用户与观察者对象的映射表
  private Hashtable<UUID,DatabaseCacheStatusObserver> observersMap=new Hashtable<UUID,DatabaseCacheStatusObserver>();
  
  /**
   * 构造函数
   * @param observer 默认的观察者,不属于任何一个监控界面用户,用于完成系统自身日志
   */
  public CompoundDatabaseCacheStatusObserver(DatabaseCacheStatusObserver observer){
    this.observers.add(observer);
  }

  /**
   * 构造函数
   * @param observers 默认的观察者,不属于任何一个监控界面用户,用于完成系统自身日志
   */
  public CompoundDatabaseCacheStatusObserver(List<DatabaseCacheStatusObserver> observers){
    this.observers.addAll(observers);
  }
  
  /**
   *  绑定一个监控界面用户以及其观察者
   * @param token 用户访问令牌
   * @param observer 观察者
   * @return 操作是否成功
   */
  public boolean attachObserver(ControllerUserToken token,DatabaseCacheStatusObserver observer){
    synchronized(observers){
      if(observers.contains(observer))
        return false;
      if(observersMap.contains(token.sid))
        return false;
      observers.add(observer);
      observersMap.put(token.sid, observer);
      observer.reloadBlackList(blacklist.get());
      observer.incomingMessageQueueSize(incomingMessageQueueSize.get());
      observer.messageQueueSize(messageQueueSize.get());
      observer.numberOfFailedMessages(numberOfFailedMessages.get());
      observer.numberOfSuccessfulMessages(numberOfSuccessfulMessages.get());
      observer.statusQueueSize(statusQueueSize.get());
      return true;
    }
  }
  
  /**
   * 解除一个监控界面用户以及其观察者的绑定
   * @param token 用户访问令牌
   * @return 操作是否成功
   */
  public boolean detachObserver(ControllerUserToken token){
    synchronized(observers){
      DatabaseCacheStatusObserver observer=observersMap.get(token.sid);
      if(observer==null)
        return false;
      observers.remove(observer);
      observersMap.remove(token.sid);
      return true;
    }
  }
  
  /**
   *  过滤掉一条短消息
   * @param message 被过滤掉的短消息
   */
  public void blockMessage(TextMessage message){
    synchronized(observers){
      for(DatabaseCacheStatusObserver observer:observers){
        observer.blockMessage(message);
      }
    }
  }

  /**
   * 设置新的消息队列尺寸
   * @param size 新尺寸
   */
  public void messageQueueSize(int size) {
    synchronized(observers){
      for(DatabaseCacheStatusObserver observer:observers){
        observer.messageQueueSize(size);
      }
      messageQueueSize.set(size);
    }
  }

  /**
   * 从数据库中取出一条新待发短消息
   * @param message 新待发短消息
   */
  public void newMessage(TextMessage message) {
    synchronized(observers){
      for(DatabaseCacheStatusObserver observer:observers){
        observer.newMessage(message);
      }
    }
  }

  /**
   * 设置累计发送失败的短信息条数
   * @param number 累计发送失败的短信息条数
   */
  public void numberOfFailedMessages(int number) {
    synchronized(observers){
      for(DatabaseCacheStatusObserver observer:observers){
        observer.numberOfFailedMessages(number);
      }
      numberOfFailedMessages.set(number);
    }
  }

  /**
   * 设置累计发送成功的短信息条数
   * @param number 累计发送成功的短信息条数
   */
  public void numberOfSuccessfulMessages(int number) {
    synchronized(observers){
      for(DatabaseCacheStatusObserver observer:observers){
        observer.numberOfSuccessfulMessages(number);
      }
      numberOfSuccessfulMessages.set(number);
    }
  }

  /**
   *  重新读取黑名单
   * @param list 黑名单
   */
  public void reloadBlackList(BlackList list) {
    synchronized(observers){
      for(DatabaseCacheStatusObserver observer:observers){
        observer.reloadBlackList(list);
      }
      blacklist.set(list);
    }
  }

  /**
   * 设置新的状态更新等待队列尺寸
   * @param size 新尺寸
   */
  public void statusQueueSize(int size) {
    synchronized(observers){
      for(DatabaseCacheStatusObserver observer:observers){
        observer.statusQueueSize(size);
      }
      statusQueueSize.set(size);
    }
  }

  /**
   * 将一条短消息的状态更新回写回数据库
   * @param message 状态更新的短消息
   */
  public void updateMessageStatus(TextMessage message) {
    synchronized(observers){
      for(DatabaseCacheStatusObserver observer:observers){
        observer.updateMessageStatus(message);
      }
    }
  }

  /**
   *  有一条状态更新的短信被添加到待更新请求队列
   * @param message 状态更新的短信
   */
  public void newMessageStatus(TextMessage message) {
    synchronized(observers){
      for(DatabaseCacheStatusObserver observer:observers){
        observer.newMessageStatus(message);
      }
    }
  }

  /**
   *  记载调试日志信息
   * @param log 调试日志信息
   */
  public void debug(String log) {
    synchronized(observers){
      for(DatabaseCacheStatusObserver observer:observers){
        observer.debug(log);
      }
    }
  }

  /**
   *  记载错误信息
   * @param log 错误信息
   */
  public void error(String log) {
    synchronized(observers){
      for(DatabaseCacheStatusObserver observer:observers){
        observer.error(log);
      }
    }
  }

  /**
   *  记载致命错误信息
   * @param log 致命错误信息
   */
  public void fatal(String log) {
    synchronized(observers){
      for(DatabaseCacheStatusObserver observer:observers){
        observer.fatal(log);
      }
    }
  }

  /**
   *  记载一般日志信息
   * @param log 一般日志信息
   */
  public void info(String log) {
    synchronized(observers){
      for(DatabaseCacheStatusObserver observer:observers){
        observer.info(log);
      }
    }
  }

  /**
   * 取得自上次调用此方法后被过滤掉的短信息内容日志
   * @return 自上次调用此方法后被过滤掉的短信息内容日志
   */
  public List<TextMessageLog> getBlockMessageLog(ControllerUserToken token){
    synchronized(observers){
      DatabaseCacheStatusObserver observer=observersMap.get(token.sid);
      if(observer!=null){
        return observer.getBlockMessageLog();
      }
      return null;
    }
  }

  /**
   * 取得自上次调用此方法后从数据库中取出的待发短信息日志
   * @return 自上次调用此方法后从数据库中取出的待发短信息日志
   */
  public List<TextMessageLog> getNewMessageLog(ControllerUserToken token){
    synchronized(observers){
      DatabaseCacheStatusObserver observer=observersMap.get(token.sid);
      if(observer!=null){
        return observer.getNewMessageLog();
      }
      return null;
    }
  }
  
  /**
   * 取得自上次调用此方法后被过短信息状态被回写回数据库的日志
   * @return 自上次调用此方法后被过短信息状态被回写回数据库的日志
   */
  public List<TextMessageLog> getUpdateMessageLog(ControllerUserToken token){
    synchronized(observers){
      DatabaseCacheStatusObserver observer=observersMap.get(token.sid);
      if(observer!=null){
        return observer.getUpdateMessageLog();
      }
      return null;
    }
  }

  /**
   * 取得自上次调用此方法后新增短信息状态更新请求的日志
   * @return 自上次调用此方法后新增短信息状态更新请求的日志
   */
  public List<TextMessageLog> getNewMessageStatusLog(ControllerUserToken token){
    synchronized(observers){
      DatabaseCacheStatusObserver observer=observersMap.get(token.sid);
      if(observer!=null){
        return observer.getNewMessageStatusLog();
      }
      return null;
    }
  }

  /**
   * 取得自上次调用此方法后所有黑名单更新日志
   * @return 自上次调用此方法后所有黑名单更新日志
   */
  public List<BlackListLog> getBlackListLog(ControllerUserToken token){
    synchronized(observers){
      DatabaseCacheStatusObserver observer=observersMap.get(token.sid);
      if(observer!=null){
        return observer.getBlackListLog();
      }
      return null;
    }
  }
  
  /**
   * 取得待发消息队列尺寸
   * @param token 用户访问令牌
   * @return 待发消息队列尺寸
   */
  public int getMessageQueueSize(ControllerUserToken token){
    synchronized(observers){
      DatabaseCacheStatusObserver observer=observersMap.get(token.sid);
      if(observer!=null){
        return observer.getMessageQueueSize();
      }
      return 0;
    }
  }
  
  /**
   * 取得累计发送失败短信息数目
   * @param token 用户访问令牌
   * @return 累计发送失败短信息数目
   */
  public int getNumberOfFailedMessages(ControllerUserToken token){
    synchronized(observers){
      DatabaseCacheStatusObserver observer=observersMap.get(token.sid);
      if(observer!=null){
        return observer.getNumberOfFailedMessages();
      }
      return 0;
    }
  }
  
  /**
   * 取得累计发送成功短信息数目
   * @param token 用户访问令牌
   * @return 累计发送成功短信息数目
   */
  public int getNumberOfSuccessfulMessages(ControllerUserToken token){
    synchronized(observers){
      DatabaseCacheStatusObserver observer=observersMap.get(token.sid);
      if(observer!=null){
        return observer.getNumberOfSuccessfulMessages();
      }
      return 0;
    }
  }
  
  /**
   * 取得状态更新回写队列尺寸
   * @param token 用户访问令牌
   * @return 状态更新回写队列尺寸
   */
  public int getStatusQueueSize(ControllerUserToken token){
    synchronized(observers){
      DatabaseCacheStatusObserver observer=observersMap.get(token.sid);
      if(observer!=null){
        return observer.getStatusQueueSize();
      }
      return 0;
    }
  }

  /**
   *  取得自上次调用此方发以后新增的所有日志信息
   * @return 自上次调用此方发以后新增的所有日志信息
   */
  public List<GenericLog> getGenericLog(ControllerUserToken token){
    synchronized(observers){
      DatabaseCacheStatusObserver observer=observersMap.get(token.sid);
      if(observer!=null){
        return observer.getGenericLog();
      }
      return null;
    }
  }

  public int getIncomingMessageQueueSize(ControllerUserToken token) {
    synchronized(observers){
      DatabaseCacheStatusObserver observer=observersMap.get(token.sid);
      if(observer!=null){
        return observer.getIncomingMessageQueueSize();
      }
      return 0;
    }
  }

  public List<TextMessageLog> getNewIncomingMessageLog(ControllerUserToken token) {
    synchronized(observers){
      DatabaseCacheStatusObserver observer=observersMap.get(token.sid);
      if(observer!=null){
        return observer.getNewIncomingMessageLog();
      }
      return null;
    }
  }

  public void incomingMessageQueueSize(int size) {
    synchronized(observers){
      for(DatabaseCacheStatusObserver observer:observers){
        observer.incomingMessageQueueSize(size);
      }
    }
  }

  public void newIncomingMessage(TextMessage message) {
    synchronized(observers){
      for(DatabaseCacheStatusObserver observer:observers){
        observer.newIncomingMessage(message);
      }
    }
  }

  /**
   *  空方法不要使用
   *  @deprecated
   */
  public List<BlackListLog> getBlackListLog() {
    return null;
  }

  /**
   *  空方法不要使用
   *  @deprecated
   */
  public List<TextMessageLog> getBlockMessageLog() {
    return null;
  }

  /**
   *  空方法不要使用
   *  @deprecated
   */
  public int getMessageQueueSize() {
    return 0;
  }

  /**
   *  空方法不要使用
   *  @deprecated
   */
  public List<TextMessageLog> getNewMessageLog() {
    return null;
  }

  /**
   *  空方法不要使用
   *  @deprecated
   */
  public List<TextMessageLog> getNewMessageStatusLog() {
    return null;
  }

  /**
   *  空方法不要使用
   *  @deprecated
   */
  public int getNumberOfFailedMessages() {
    return 0;
  }

  /**
   *  空方法不要使用
   *  @deprecated
   */
  public int getNumberOfSuccessfulMessages() {
    return 0;
  }

  /**
   *  空方法不要使用
   *  @deprecated
   */
  public int getStatusQueueSize() {
    return 0;
  }

  /**
   *  空方法不要使用
   *  @deprecated
   */
  public List<TextMessageLog> getUpdateMessageLog() {
    return null;
  }

  /**
   *  空方法不要使用
   *  @deprecated
   */
  public List<GenericLog> getGenericLog() {
    return null;
  }

  /**
   *  空方法不要使用
   *  @deprecated
   */
  public int getIncomingMessageQueueSize() {
    return 0;
  }

  /**
   *  空方法不要使用
   *  @deprecated
   */
  public List<TextMessageLog> getNewIncomingMessageLog() {
    return null;
  }

  /**
   *  空方法不要使用
   *  @deprecated
   */
  public void warn(String log) {
  }

  /**
   *  空方法不要使用
   *  @deprecated
   */
  public void debug(Exception ex) {
  }

  /**
   *  空方法不要使用
   *  @deprecated
   */
  public void error(Exception ex) {
  }

  /**
   *  空方法不要使用
   *  @deprecated
   */
  public void fatal(Exception ex) {
  }

  /**
   *  空方法不要使用
   *  @deprecated
   */
  public void info(Exception ex) {
  }

  /**
   *  空方法不要使用
   *  @deprecated
   */
  public void warn(Exception ex) {
  }
}