package com.ctrip.xpipe.redis.console.service.impl;

import com.ctrip.xpipe.redis.console.model.KeepercontainerTbl;
import com.ctrip.xpipe.redis.console.model.RedisTbl;
import com.ctrip.xpipe.redis.console.model.RedisTblDao;
import com.ctrip.xpipe.redis.console.service.*;
import com.ctrip.xpipe.redis.core.protocal.RedisProtocol;
import com.ctrip.xpipe.spring.AbstractProfile;
import com.ctrip.xpipe.spring.AbstractSpringConfigContext;
import com.ctrip.xpipe.utils.VisibleForTesting;
import com.google.common.collect.Maps;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.util.*;
import java.util.concurrent.ExecutorService;
import java.util.function.BiPredicate;

/**
 * @author wenchao.meng
 *         <p>
 *         Jul 17, 2017
 */
@Component
public class DefaultKeeperAdvancedService extends AbstractConsoleService<RedisTblDao> implements KeeperAdvancedService {

  @Autowired
  private KeepercontainerService keepercontainerService;

  @Autowired
  private RedisService redisService;

  @Resource(name = AbstractSpringConfigContext.GLOBAL_EXECUTOR)
  ExecutorService executor;

  @Override
  public List<KeeperBasicInfo> findBestKeepers(String dcName, String clusterName) {
    return findBestKeepers(dcName, RedisProtocol.REDIS_PORT_DEFAULT, (host, port) -> true, clusterName);
  }

  @Override
  public List<KeeperBasicInfo> findBestKeepers(String dcName, int beginPort, BiPredicate keeperGood,
      String clusterName) {
    return findBestKeepers(dcName, beginPort, keeperGood, clusterName, 2);
  }

  public List<KeeperBasicInfo> findBestKeepers(String dcName, int beginPort, BiPredicate<String, Integer> keeperGood,
      String clusterName, int returnCount) {

    List<KeeperBasicInfo> result = new LinkedList<>();

    List<KeepercontainerTbl> keepercontainerTbls =
        keepercontainerService.findBestKeeperContainersByDcCluster(dcName, clusterName);
    if (keepercontainerTbls.size() < returnCount) {
      throw new IllegalStateException(
          "Organization keepers size:" + keepercontainerTbls.size() + ", but we need:" + returnCount);
    }

    fillInResult(keepercontainerTbls, result, beginPort, keeperGood, returnCount);
    return result;

  }


  private void fillInResult(List<KeepercontainerTbl> keeperCount, List<KeeperBasicInfo> result, int beginPort,
      BiPredicate<String, Integer> keeperGood, int returnCount) {

    Map<String, Set<Integer>> ipAndPorts = getIpAndPortsWithSameIpAsKC(keeperCount);
    // find available port
    for (int i = 0; i < returnCount; i++) {

      KeepercontainerTbl keepercontainerTbl = keeperCount.get(i);

      KeeperBasicInfo keeperSelected = new KeeperBasicInfo();

      keeperSelected.setKeeperContainerId(keepercontainerTbl.getKeepercontainerId());
      keeperSelected.setHost(keepercontainerTbl.getKeepercontainerIp());

      int port = findAvailablePort(keepercontainerTbl, beginPort, keeperGood, result, ipAndPorts);

      keeperSelected.setPort(port);
      result.add(keeperSelected);
    }
  }

  @VisibleForTesting
  int findAvailablePort(KeepercontainerTbl keepercontainerTbl, int beginPort,
                        BiPredicate<String, Integer> keeperGood, List<KeeperBasicInfo> result, Map<String, Set<Integer>> ipAndPorts) {

    int port = beginPort;
    String ip = keepercontainerTbl.getKeepercontainerIp();

    Set<Integer> existingPorts = ipAndPorts.get(ip);

    for (;; port++) {

      if (alreadySelected(ip, port, result)) {
        continue;
      }

      if (!(keeperGood.test(ip, port))) {
        continue;
      }
      if (existingPorts.contains(port)) {
        continue;
      }

      break;
    }

    return port;
  }

  private boolean alreadySelected(String ip, int port, List<KeeperBasicInfo> result) {

    for (KeeperBasicInfo keeperSelected : result) {
      if (keeperSelected.getHost().equalsIgnoreCase(ip) && keeperSelected.getPort() == port) {
        return true;
      }
    }
    return false;
  }


  private Map<String, Set<Integer>> getIpAndPortsWithSameIpAsKC(List<KeepercontainerTbl> keeperCount) {

    Map<String, Set<Integer>> map = Maps.newHashMap();
    keeperCount.forEach(kc -> map.putIfAbsent(kc.getKeepercontainerIp(), new HashSet<Integer>()));

    for(Map.Entry<String, Set<Integer>> entry : map.entrySet()) {

      String ip = entry.getKey();
      Set<Integer> existingPorts = entry.getValue();
      executor.submit(new Runnable() {
        @Override
        public void run() {
          List<RedisTbl> redisWithSameIP = redisService.findAllRedisWithSameIP(ip);
          redisWithSameIP.forEach(redisTbl -> existingPorts.add(redisTbl.getRedisPort()));
        }
      });
    }
    return map;
  }
}

@@ -718,24 +718,108 @@
 t.Execut
-orServic
+ionException;%0Aimport java.util.concurrent.ExecutorService;%0Aimport java.util.concurrent.Futur
 e;%0Aimpor
@@ -4376,24 +4376,80 @@
 eger%3E()));%0A%0A
+    List%3CFuture%3E futures = new ArrayList%3C%3E(map.size());%0A
     for(Map.
@@ -4498,24 +4498,24 @@
 trySet()) %7B%0A
-
 %0A      Strin
@@ -4595,16 +4595,32 @@
 );%0A     
+ Future future =
  executo
@@ -4872,24 +4872,24 @@
 ;%0A        %7D%0A
-
       %7D);%0A  
@@ -4886,16 +4886,452 @@
     %7D);%0A
+      futures.add(future);%0A    %7D%0A    for(Future future : futures) %7B%0A      try %7B%0A        future.get();%0A      %7D catch (InterruptedException ignore) %7B%0A      %7D catch (ExecutionException e) %7B%0A        for(Future futureToCancel : futures) %7B%0A          if(!futureToCancel.isDone() %7C%7C !futureToCancel.isCancelled()) %7B%0A            futureToCancel.cancel(true);%0A          %7D%0A        %7D%0A        return getIpAndPortsWithSameIpAsKC(keeperCount);%0A      %7D%0A
     %7D%0A  

add future wait logic