content: /*
   * Copyright (C) 2010-2101 Alibaba Group Holding Limited.
   *
   * Licensed under the Apache License, Version 2.0 (the "License");
   * you may not use this file except in compliance with the License.
   * You may obtain a copy of the License at
   *
   *     http://www.apache.org/licenses/LICENSE-2.0
   *
   * Unless required by applicable law or agreed to in writing, software
   * distributed under the License is distributed on an "AS IS" BASIS,
   * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
   * See the License for the specific language governing permissions and
   * limitations under the License.
   */
  
  package com.alibaba.otter.shared.arbitrate.impl.setl.monitor;
  
  import java.util.ArrayList;
  import java.util.Collections;
  import java.util.List;
  import java.util.concurrent.ExecutorService;
  
  import org.I0Itec.zkclient.IZkDataListener;
  import org.I0Itec.zkclient.exception.ZkException;
  import org.I0Itec.zkclient.exception.ZkNoNodeException;
  import org.apache.commons.lang.ClassUtils;
  import org.slf4j.Logger;
  import org.slf4j.LoggerFactory;
  import org.slf4j.MDC;
  
  import com.alibaba.otter.shared.arbitrate.impl.ArbitrateConstants;
  import com.alibaba.otter.shared.arbitrate.impl.config.ArbitrateConfigUtils;
  import com.alibaba.otter.shared.arbitrate.impl.setl.ArbitrateLifeCycle;
  import com.alibaba.otter.shared.arbitrate.impl.setl.helper.StagePathUtils;
  import com.alibaba.otter.shared.arbitrate.impl.setl.monitor.listener.PermitListener;
  import com.alibaba.otter.shared.arbitrate.impl.zookeeper.ZooKeeperClient;
  import com.alibaba.otter.shared.arbitrate.model.MainStemEventData;
  import com.alibaba.otter.shared.common.model.config.channel.ChannelStatus;
  import com.alibaba.otter.shared.common.utils.JsonUtils;
  import com.alibaba.otter.shared.common.utils.lock.BooleanMutex;
  import com.alibaba.otter.shared.common.utils.zookeeper.ZkClientx;
  
  /**
   * 同步任务状态的监控
   * 
   * <pre>
   * 监控数据内容：
   * 1. channel的status状态
   * 2. 当前pipeline的mainStem状态 &　反向同步的pipeline的mainStem状态
   * </pre>
   * 
   * @author jianghang
   */
  public class PermitMonitor extends ArbitrateLifeCycle implements Monitor {
  
      private static final Logger      logger                 = LoggerFactory.getLogger(PermitMonitor.class);
  
      private ZkClientx                zookeeper              = ZooKeeperClient.getInstance();
      private ChannelStatus            channelStatus          = ChannelStatus.STOP;                                           // 标识channel的状态
      private MainStemEventData.Status mainStemStatus         = MainStemEventData.Status.TAKEING;                             // 当前pipeline的mainStem状态
      private MainStemEventData.Status oppositeMainStemStatus = MainStemEventData.Status.TAKEING;                             // 反方向的pipeline的mainStem状态
  
      private ExecutorService          arbitrateExecutor;
      private BooleanMutex             permitMutex            = new BooleanMutex(false);                                      // 控制器
      private BooleanMutex             channelMutex           = new BooleanMutex(false);
      private List<PermitListener>     listeners              = Collections.synchronizedList(new ArrayList<PermitListener>());
      private volatile boolean         existOpposite          = false;
      private IZkDataListener          channelDataListener;
      private IZkDataListener          mainstemDataListener;
      private IZkDataListener          oppositeMainstemDataListener;
  
      public PermitMonitor(Long pipelineId){
          super(pipelineId);
          existOpposite = (ArbitrateConfigUtils.getOppositePipeline(getPipelineId()) != null);
          // 开始同步
          channelDataListener = new IZkDataListener() {
  
              public void handleDataChange(String dataPath, Object data) throws Exception {
                  initChannelStatus((byte[]) data);
              }
  
              public void handleDataDeleted(String dataPath) throws Exception {
                  channelStatus = ChannelStatus.STOP;
                  permitSem();
              }
          };
          String path = StagePathUtils.getChannel(getPipelineId());
          zookeeper.subscribeDataChanges(path, channelDataListener);
  
          mainstemDataListener = new IZkDataListener() {
  
              public void handleDataChange(String dataPath, Object data) throws Exception {
                  initMainStemStatus((byte[]) data);
              }
  
              public void handleDataDeleted(String dataPath) throws Exception {
                  // mainstem节点挂了后，状态直接修改为taking
                  mainStemStatus = MainStemEventData.Status.TAKEING;
                  permitSem();
              }
          };
  
          path = StagePathUtils.getMainStem(getPipelineId());
          zookeeper.subscribeDataChanges(path, mainstemDataListener);
  
          initChannelStatus();
          initMainStemStatus();
          // syncChannelStatus();
          // syncMainStemStatus();
          if (existOpposite) {
              oppositeMainstemDataListener = new IZkDataListener() {
  
                  public void handleDataChange(String dataPath, Object data) throws Exception {
                      initOppositeMainStemStatus((byte[]) data);
                  }
  
                  public void handleDataDeleted(String dataPath) throws Exception {
                      // mainstem节点挂了后，状态直接修改为taking
                      oppositeMainStemStatus = MainStemEventData.Status.TAKEING;
                      permitSem();
                  }
              };
  
              path = StagePathUtils.getOppositeMainStem(getPipelineId());
              zookeeper.subscribeDataChanges(path, oppositeMainstemDataListener);
              initOppositeMainStemStatus();
              // syncOppositeMainStemStatus();
          }
          MonitorScheduler.register(this);
      }
  
      public void reload() {
          if (logger.isDebugEnabled()) {
              logger.debug("## reload Permit pipeline[{}]", getPipelineId());
          }
  
          try {
              initChannelStatus();
          } catch (Exception e) {// 处理下异常
          }
  
          try {
              initMainStemStatus();
          } catch (Exception e) {// 处理下异常
          }
  
          boolean prev = existOpposite;
          existOpposite = (ArbitrateConfigUtils.getOppositePipeline(getPipelineId()) != null);
          if (existOpposite) {
              if (prev == false) {
                  // syncOppositeMainStemStatus();// 是个变化的过程，开启反向同步
                  String path = StagePathUtils.getOppositeMainStem(getPipelineId());
                  zookeeper.subscribeDataChanges(path, oppositeMainstemDataListener);
              }
  
              try {
                  initOppositeMainStemStatus();
              } catch (Exception e) {// 处理下异常
              }
          }
  
      }
  
      public void destory() {
          super.destory();
  
          if (logger.isDebugEnabled()) {
              logger.debug("## destory Permit pipeline[{}]", getPipelineId());
          }
          String path = StagePathUtils.getChannel(getPipelineId());
          zookeeper.unsubscribeDataChanges(path, channelDataListener);
  
          path = StagePathUtils.getMainStem(getPipelineId());
          zookeeper.unsubscribeDataChanges(path, mainstemDataListener);
  
          if (existOpposite) {
              path = StagePathUtils.getOppositeMainStem(getPipelineId());
              zookeeper.unsubscribeDataChanges(path, oppositeMainstemDataListener);
          }
  
          MonitorScheduler.unRegister(this);
      }
  
      /**
       * 查询是否允许授权处理，非阻塞
       */
      public boolean isPermit() {
          return isPermit(false);
      }
  
      /**
       * 查询是否允许授权处理，非阻塞，指定是否强制刷新
       */
      public boolean isPermit(boolean reload) {
          if (reload) {// 判断是否需要重新reload
              reload();
          }
  
          boolean result = channelStatus.isStart() && mainStemStatus.isOverTake();
          if (existOpposite) {// 判断是否存在反向同步
              result &= oppositeMainStemStatus.isOverTake();
          }
  
          return result;
      }
  
      /**
       * 查询对应的channel授权状态
       */
      public ChannelStatus getChannelPermit() {
          return getChannelPermit(false);
      }
  
      /**
       * 查询对应的mainstem授权状态
       */
      public MainStemEventData.Status getMainStemPermit() {
          return getMainStemPermit(false);
      }
  
      /**
       * 查询对应的channel授权状态，指定是否强制刷新
       */
      public ChannelStatus getChannelPermit(boolean reload) {
          if (reload) {
              initChannelStatus();
          }
  
          return channelStatus;
      }
  
      /**
       * 查询对应的mainstem授权状态，指定是否强制刷新
       */
      public MainStemEventData.Status getMainStemPermit(boolean reload) {
          if (reload) {
              initMainStemStatus();
          }
  
          return mainStemStatus;
      }
  
      /**
       * 阻塞等待允许授权处理, 支持线程中断信号
       * 
       * @return
       */
      public void waitForPermit() throws InterruptedException {
          permitMutex.get();
      }
  
      /**
       * 阻塞等待允许channel的授权处理, 支持线程中断信号
       * 
       * @throws InterruptedException
       */
      public void waitForChannelPermit() throws InterruptedException {
          channelMutex.get();
      }
  
      // ================ 状态数据同步 ================
  
      private void initChannelStatus() {
          String path = null;
          try {
              path = StagePathUtils.getChannel(getPipelineId());
              byte[] bytes = zookeeper.readData(path);
              initChannelStatus(bytes);
          } catch (ZkNoNodeException e) {
              channelStatus = ChannelStatus.STOP;
              permitSem();
          } catch (ZkException e) {
              logger.error(path, e);
          }
      }
  
      private void initChannelStatus(byte[] bytes) {
          ChannelStatus newChannelStatus = JsonUtils.unmarshalFromByte(bytes, ChannelStatus.class);
  
          if (logger.isDebugEnabled()) {
              logger.debug("pipeline[{}] newChannelStatus is [{}]", getPipelineId(), newChannelStatus);
          }
  
          synchronized (this) {
              // 发生变化，才触发权限检查
              if (!newChannelStatus.equals(channelStatus)) {
                  channelStatus = newChannelStatus;
                  permitSem();
              }
          }
      }
  
      // private void syncChannelStatus() {
      // final String path = StagePathUtils.getChannel(getPipelineId());
      // try {
      // zookeeper.exists(path, new AsyncWatcher() {
      //
      // public void asyncProcess(WatchedEvent event) {
      // MDC.put(ArbitrateConstants.splitPipelineLogFileKey, String.valueOf(getPipelineId()));
      // if (isStop()) {// 如果已关闭，停止递归同步
      // return;
      // }
      // // 出现session expired/connection losscase下，会触发所有的watcher响应，同时老的watcher会继续保留，所以会导致出现多次watcher响应
      // boolean dataChanged = event.getType() == EventType.NodeDataChanged
      // || event.getType() == EventType.NodeDeleted
      // || event.getType() == EventType.NodeCreated
      // || event.getType() == EventType.NodeChildrenChanged;
      // if (dataChanged) {
      // syncChannelStatus();// 开始同步channel状态
      // }
      // }
      // });
      //
      // // 防止 watcher 错过事件
      // initChannelStatus();
      // } catch (KeeperException e) {
      // syncChannelStatus();// 开始同步channel状态
      // logger.error(path, e);
      // } catch (InterruptedException e) {
      // // ignore
      // }
      // }
  
      private void initMainStemStatus() {
          String path = null;
          try {
              path = StagePathUtils.getMainStem(getPipelineId());
              byte[] bytes = zookeeper.readData(path);
              initMainStemStatus(bytes);
          } catch (ZkNoNodeException e) {
              // mainstem节点挂了后，状态直接修改为taking
              mainStemStatus = MainStemEventData.Status.TAKEING;
              permitSem();
          } catch (ZkException e) {
              logger.error(path, e);
          }
      }
  
      private void initMainStemStatus(byte[] bytes) {
          MainStemEventData eventData = JsonUtils.unmarshalFromByte(bytes, MainStemEventData.class);
          MainStemEventData.Status newStatus = eventData.getStatus();
  
          if (logger.isDebugEnabled()) {
              logger.debug("pipeline[{}] new mainStemStatus is [{}]", getPipelineId(), newStatus);
          }
  
          synchronized (this) {
              if (!mainStemStatus.equals(newStatus)) {
                  mainStemStatus = newStatus;
                  permitSem();
              }
          }
      }
  
      // private void syncMainStemStatus() {
      // final String path = StagePathUtils.getMainStem(getPipelineId());
      // try {
      // // exists同样在data发生变化时会触发
      // zookeeper.exists(path, new AsyncWatcher() {
      //
      // public void asyncProcess(WatchedEvent event) {
      // MDC.put(ArbitrateConstants.splitPipelineLogFileKey, String.valueOf(getPipelineId()));
      // if (isStop()) {// 如果已关闭，停止递归同步
      // return;
      // }
      // // 出现session expired/connection losscase下，会触发所有的watcher响应，同时老的watcher会继续保留，所以会导致出现多次watcher响应
      // boolean dataChanged = event.getType() == EventType.NodeDataChanged
      // || event.getType() == EventType.NodeDeleted
      // || event.getType() == EventType.NodeCreated
      // || event.getType() == EventType.NodeChildrenChanged;
      // if (dataChanged) {
      // syncMainStemStatus();// 开始同步mainStem状态
      // }
      // }
      // });
      //
      // initMainStemStatus();
      // } catch (NoNodeException e) {
      // // 可能不存在对应的节点,忽略
      // } catch (KeeperException e) {
      // syncMainStemStatus();// 开始同步mainStem状态
      // logger.error(path, e);
      // } catch (InterruptedException e) {
      // // ignore
      // }
      // }
  
      private void initOppositeMainStemStatus() {
          String path = null;
          try {
              path = StagePathUtils.getOppositeMainStem(getPipelineId());
              byte[] bytes = zookeeper.readData(path);
              initOppositeMainStemStatus(bytes);
          } catch (ZkNoNodeException e) {
              // mainstem节点挂了后，状态直接修改为taking
              oppositeMainStemStatus = MainStemEventData.Status.TAKEING;
              permitSem();
          } catch (ZkException e) {
              logger.error(path, e);
          }
      }
  
      private void initOppositeMainStemStatus(byte[] bytes) {
          MainStemEventData eventData = JsonUtils.unmarshalFromByte(bytes, MainStemEventData.class);
          MainStemEventData.Status newStatus = eventData.getStatus();
  
          if (logger.isDebugEnabled()) {
              logger.debug("pipeline[{}] new oppositeMainStemStatus is [{}]", getPipelineId(), newStatus);
          }
  
          synchronized (this) {
              if (!oppositeMainStemStatus.equals(newStatus)) {
                  oppositeMainStemStatus = newStatus;
                  permitSem();
              }
          }
      }
  
      // private void syncOppositeMainStemStatus() {
      // final String path = StagePathUtils.getOppositeMainStem(getPipelineId());
      // try {
      // // exists同样在data发生变化时会触发
      // zookeeper.exists(path, new AsyncWatcher() {
      //
      // public void asyncProcess(WatchedEvent event) {
      // MDC.put(ArbitrateConstants.splitPipelineLogFileKey, String.valueOf(getPipelineId()));
      // if (existOpposite == false || isStop()) {// 如果已关闭，停止递归同步
      // return;
      // }
      // // 出现session expired/connection losscase下，会触发所有的watcher响应，同时老的watcher会继续保留，所以会导致出现多次watcher响应
      // boolean dataChanged = event.getType() == EventType.NodeDataChanged
      // || event.getType() == EventType.NodeDeleted
      // || event.getType() == EventType.NodeCreated
      // || event.getType() == EventType.NodeChildrenChanged;
      // if (dataChanged) {
      // syncOppositeMainStemStatus();// 开始同步mainStem状态
      // }
      // }
      // });
      //
      // initOppositeMainStemStatus();
      // } catch (NoNodeException e) {
      // // 可能不存在对应的opposite节点,忽略
      // } catch (KeeperException e) {
      // syncOppositeMainStemStatus();// 开始同步mainStem状态
      // logger.error(path, e);
      // } catch (InterruptedException e) {
      // // ignore
      // }
      // }
  
      /**
       * permit信号的处理
       */
      private void permitSem() {
          if (channelStatus.isStart()) {
              channelMutex.set(true);
              logger.debug("channel status is ok!");
          } else {
              channelMutex.set(false);
              logger.debug("channel status is fail!");
          }
  
          boolean permit = isPermit(false);
          if (permit == false) {
              if (logger.isDebugEnabled()) {
                  logger.debug("Permit is fail!");
              }
              // 如果未授权，则设置信号量为0
              permitMutex.set(false);
          } else {
              // 信号量+1
              if (logger.isDebugEnabled()) {
                  logger.debug("Permit is Ok!");
              }
              permitMutex.set(true);
          }
  
          processChanged(permit);// 通知下变化
      }
  
      // ======================== listener处理 ======================
  
      public void addListener(PermitListener listener) {
          if (logger.isDebugEnabled()) {
              logger.debug("## pipeline[{}] add listener [{}]", getPipelineId(),
                           ClassUtils.getShortClassName(listener.getClass()));
          }
  
          this.listeners.add(listener);
      }
  
      public void removeListener(PermitListener listener) {
          if (logger.isDebugEnabled()) {
              logger.debug("## pipeline[{}] remove listener [{}]", getPipelineId(),
                           ClassUtils.getShortClassName(listener.getClass()));
          }
  
          this.listeners.remove(listener);
      }
  
      private void processChanged(final boolean isPermit) {
          for (final PermitListener listener : listeners) {
              // 异步处理
              arbitrateExecutor.submit(new Runnable() {
  
                  public void run() {
                      MDC.put(ArbitrateConstants.splitPipelineLogFileKey, String.valueOf(getPipelineId()));
                      listener.processChanged(isPermit);
                  }
              });
          }
      }
  
      public void setArbitrateExecutor(ExecutorService arbitrateExecutor) {
          this.arbitrateExecutor = arbitrateExecutor;
      }
  
  }
  