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.manager.biz.remote.impl;
  
  import java.util.ArrayList;
  import java.util.Collections;
  import java.util.HashSet;
  import java.util.List;
  import java.util.Set;
  
  import org.apache.commons.lang.ArrayUtils;
  import org.apache.commons.lang.StringUtils;
  import org.slf4j.Logger;
  import org.slf4j.LoggerFactory;
  import org.springframework.util.Assert;
  import org.springframework.util.CollectionUtils;
  
  import com.alibaba.otter.manager.biz.common.exceptions.ManagerException;
  import com.alibaba.otter.manager.biz.config.channel.ChannelService;
  import com.alibaba.otter.manager.biz.config.datamatrix.DataMatrixService;
  import com.alibaba.otter.manager.biz.config.node.NodeService;
  import com.alibaba.otter.manager.biz.remote.ConfigRemoteService;
  import com.alibaba.otter.shared.common.model.config.channel.Channel;
  import com.alibaba.otter.shared.common.model.config.channel.ChannelStatus;
  import com.alibaba.otter.shared.common.model.config.data.DataMatrix;
  import com.alibaba.otter.shared.common.model.config.node.Node;
  import com.alibaba.otter.shared.common.model.config.pipeline.Pipeline;
  import com.alibaba.otter.shared.common.utils.JsonUtils;
  import com.alibaba.otter.shared.communication.core.CommunicationClient;
  import com.alibaba.otter.shared.communication.core.CommunicationRegistry;
  import com.alibaba.otter.shared.communication.model.config.ConfigEventType;
  import com.alibaba.otter.shared.communication.model.config.FindChannelEvent;
  import com.alibaba.otter.shared.communication.model.config.FindMediaEvent;
  import com.alibaba.otter.shared.communication.model.config.FindNodeEvent;
  import com.alibaba.otter.shared.communication.model.config.FindTaskEvent;
  import com.alibaba.otter.shared.communication.model.config.NotifyChannelEvent;
  
  /**
   * Config的remote接口处理
   * 
   * @author jianghang 2011-10-21 下午02:53:53
   * @version 4.0.0
   */
  public class ConfigRemoteServiceImpl implements ConfigRemoteService {
  
      private static final Logger logger = LoggerFactory.getLogger(ConfigRemoteServiceImpl.class);
      private CommunicationClient communicationClient;
      private ChannelService      channelService;
      private NodeService         nodeService;
      private DataMatrixService   dataMatrixService;
  
      public ConfigRemoteServiceImpl(){
          // 注册一下事件处理
          CommunicationRegistry.regist(ConfigEventType.findChannel, this);
          CommunicationRegistry.regist(ConfigEventType.findNode, this);
          CommunicationRegistry.regist(ConfigEventType.findTask, this);
          CommunicationRegistry.regist(ConfigEventType.findMedia, this);
      }
  
      public boolean notifyChannel(final Channel channel) {
          Assert.notNull(channel);
          // 获取所有的Node节点
          NotifyChannelEvent event = new NotifyChannelEvent();
          event.setChannel(channel);
  
          Set<String> addrsSet = new HashSet<String>();
  
          // 组装当前otter所有的存活的node节点
          // List<Node> nodes = nodeService.listAll();
          // for (Node node : nodes) {
          // if (node.getStatus().isStart() &&
          // StringUtils.isNotEmpty(node.getIp()) && node.getPort() != 0) {
          // final String addr = node.getIp() + ":" + node.getPort();
          // addrsList.add(addr);
          // }
          // }
  
          // 组装当前pipeline下的存活的node节点
          for (Pipeline pipeline : channel.getPipelines()) {
              List<Node> nodes = new ArrayList<Node>();
              nodes.addAll(pipeline.getSelectNodes());
              nodes.addAll(pipeline.getExtractNodes());
              nodes.addAll(pipeline.getLoadNodes());
              for (Node node : nodes) {
                  if (node.getStatus().isStart() && StringUtils.isNotEmpty(node.getIp()) && node.getPort() != 0) {
                      String addr = node.getIp() + ":" + node.getPort();
                      if (node.getParameters().getUseExternalIp()) {
                          addr = node.getParameters().getExternalIp() + ":" + node.getPort();
                      }
                      addrsSet.add(addr);
                  }
              }
          }
  
          List<String> addrsList = new ArrayList<String>(addrsSet);
          if (CollectionUtils.isEmpty(addrsList) && channel.getStatus().isStart()) {
              throw new ManagerException("no live node for notifyChannel");
          } else if (CollectionUtils.isEmpty(addrsList)) {
              // 针对关闭操作，可直接处理
              return true;
          } else {
              Collections.shuffle(addrsList);// 做一下随机，避免每次选择的机器都是同一台
              try {
                  String[] addrs = addrsList.toArray(new String[addrsList.size()]);
                  List<Boolean> result = (List<Boolean>) communicationClient.call(addrs, event); // 推送配置
                  logger.info("## notifyChannel to [{}] channel[{}] result[{}]",
                      new Object[] { ArrayUtils.toString(addrs), channel.toString(), result });
  
                  boolean flag = true;
                  for (Boolean f : result) {
                      flag &= f;
                  }
  
                  return flag;
              } catch (Exception e) {
                  logger.error("## notifyChannel error!", e);
                  throw new ManagerException(e);
              }
          }
      }
  
      /**
       * 根据对应的工作节点机器id，获取相关的channel任务
       */
      public Channel onFindChannel(FindChannelEvent event) {
          Assert.notNull(event);
          Long channelId = event.getChannelId();
          Long pipelineId = event.getPipelineId();
          Channel channel = null;
          if (channelId != null) {
              channel = channelService.findById(channelId);
          } else {
              Assert.notNull(pipelineId);
              channel = channelService.findByPipelineId(pipelineId);
          }
  
          return channel;
      }
  
      public Node onFindNode(FindNodeEvent event) {
          Assert.notNull(event);
          Assert.notNull(event.getNid());
          return nodeService.findById(event.getNid());
      }
  
      public List<Channel> onFindTask(FindTaskEvent event) {
          Assert.notNull(event);
          Assert.notNull(event.getNid());
          // 同时查询start/pause状态的同步任务，因为在发布时重启jvm刚好在执行stopNode的restart指令，此时channel处于pause状态，丢失了任务命令
          return channelService.listByNodeId(event.getNid(), ChannelStatus.START, ChannelStatus.PAUSE);
      }
  
      public String onFindMedia(FindMediaEvent event) {
          Assert.notNull(event);
          Assert.notNull(event.getDataId());
          DataMatrix matrix = dataMatrixService.findByGroupKey(event.getDataId());
          return JsonUtils.marshalToString(matrix);
      }
  
      // =============== setter / getter ===================
  
      public void setCommunicationClient(CommunicationClient communicationClient) {
          this.communicationClient = communicationClient;
      }
  
      public void setChannelService(ChannelService channelService) {
          this.channelService = channelService;
      }
  
      public void setNodeService(NodeService nodeService) {
          this.nodeService = nodeService;
      }
  
      public void setDataMatrixService(DataMatrixService dataMatrixService) {
          this.dataMatrixService = dataMatrixService;
      }
  
  }
  