package com.ctrip.framework.apollo.internals;

 import com.google.common.base.Joiner;

 import com.google.common.base.Strings;

 import com.google.common.collect.Lists;

 import com.google.common.collect.Maps;

 import com.google.common.escape.Escaper;

 import com.google.common.net.UrlEscapers;

 import com.google.common.util.concurrent.RateLimiter;

 import com.ctrip.framework.apollo.Apollo;

 import com.ctrip.framework.apollo.core.ConfigConsts;

 import com.ctrip.framework.apollo.core.dto.ApolloConfig;

 import com.ctrip.framework.apollo.core.dto.ApolloConfigNotification;

 import com.ctrip.framework.apollo.core.dto.ServiceDTO;

 import com.ctrip.framework.apollo.core.schedule.ExponentialSchedulePolicy;

 import com.ctrip.framework.apollo.core.schedule.SchedulePolicy;

 import com.ctrip.framework.apollo.core.utils.ApolloThreadFactory;

 import com.ctrip.framework.apollo.exceptions.ApolloConfigException;

 import com.ctrip.framework.apollo.util.ConfigUtil;

 import com.ctrip.framework.apollo.util.ExceptionUtil;

 import com.ctrip.framework.apollo.util.http.HttpRequest;

 import com.ctrip.framework.apollo.util.http.HttpResponse;

 import com.ctrip.framework.apollo.util.http.HttpUtil;

 import com.dianping.cat.Cat;

 import com.dianping.cat.message.Message;

 import com.dianping.cat.message.Transaction;

 import org.codehaus.plexus.PlexusContainer;

 import org.codehaus.plexus.component.repository.exception.ComponentLookupException;

 import org.slf4j.Logger;

 import org.slf4j.LoggerFactory;

 import org.unidal.lookup.ContainerLoader;

 import java.util.Collections;

 import java.util.List;

 import java.util.Map;

 import java.util.Properties;

 import java.util.Random;

 import java.util.concurrent.ExecutorService;

 import java.util.concurrent.Executors;

 import java.util.concurrent.ScheduledExecutorService;

 import java.util.concurrent.TimeUnit;

 import java.util.concurrent.atomic.AtomicBoolean;

 import java.util.concurrent.atomic.AtomicReference;

 /**

  * @author Jason Song(song_s@ctrip.com)

  */

 public class RemoteConfigRepository extends AbstractConfigRepository {

   private static final Logger logger = LoggerFactory.getLogger(RemoteConfigRepository.class);

   private static final Joiner STRING_JOINER = Joiner.on(ConfigConsts.CLUSTER_NAMESPACE_SEPARATOR);

   private static final Joiner.MapJoiner MAP_JOINER = Joiner.on("&").withKeyValueSeparator("=");

   private PlexusContainer m_container;

   private final ConfigServiceLocator m_serviceLocator;

   private final HttpUtil m_httpUtil;

   private final ConfigUtil m_configUtil;

   private volatile AtomicReference<ApolloConfig> m_configCache;

   private final String m_namespace;

   private final static ScheduledExecutorService m_executorService;

   private final ExecutorService m_longPollingService;

   private final AtomicBoolean m_longPollingStopped;

   private SchedulePolicy m_longPollFailSchedulePolicyInSecond;

   private AtomicReference<ServiceDTO> m_longPollServiceDto;

   private AtomicReference<ApolloConfigNotification> m_longPollResult;

   private RateLimiter m_longPollRateLimiter;

   private RateLimiter m_loadConfigRateLimiter;

   private static final Escaper pathEscaper = UrlEscapers.urlPathSegmentEscaper();

   private static final Escaper queryParamEscaper = UrlEscapers.urlFormParameterEscaper();

   static {

     m_executorService = Executors.newScheduledThreadPool(1,

         ApolloThreadFactory.create("RemoteConfigRepository", true));

   }

   /**

    * Constructor.

    *

    * @param namespace the namespace

    */

   public RemoteConfigRepository(String namespace) {

     m_namespace = namespace;

     m_configCache = new AtomicReference<>();

     m_container = ContainerLoader.getDefaultContainer();

     try {

       m_configUtil = m_container.lookup(ConfigUtil.class);

       m_httpUtil = m_container.lookup(HttpUtil.class);

       m_serviceLocator = m_container.lookup(ConfigServiceLocator.class);

     } catch (ComponentLookupException ex) {

       Cat.logError(ex);

       throw new ApolloConfigException("Unable to load component!", ex);

     }

     m_longPollFailSchedulePolicyInSecond = new ExponentialSchedulePolicy(1, 120); //in second

     m_longPollingStopped = new AtomicBoolean(false);

     m_longPollingService = Executors.newSingleThreadExecutor(

         ApolloThreadFactory.create("RemoteConfigRepository-LongPolling", true));

     m_longPollServiceDto = new AtomicReference<>();

     m_longPollResult = new AtomicReference<>();

     m_longPollRateLimiter = RateLimiter.create(m_configUtil.getLongPollQPS());

     m_loadConfigRateLimiter = RateLimiter.create(m_configUtil.getLoadConfigQPS());

     this.trySync();

     this.schedulePeriodicRefresh();

     this.scheduleLongPollingRefresh();

   }

   @Override

   public Properties getConfig() {

     if (m_configCache.get() == null) {

       this.sync();

     }

     return transformApolloConfigToProperties(m_configCache.get());

   }

   @Override

   public void setUpstreamRepository(ConfigRepository upstreamConfigRepository) {

     //remote config doesn't need upstream

   }

   private void schedulePeriodicRefresh() {

     logger.debug("Schedule periodic refresh with interval: {} {}",

         m_configUtil.getRefreshInterval(), m_configUtil.getRefreshIntervalTimeUnit());

     this.m_executorService.scheduleAtFixedRate(

         new Runnable() {

           @Override

           public void run() {

             Cat.logEvent("Apollo.ConfigService", String.format("periodicRefresh: %s", m_namespace));

             logger.debug("refresh config for namespace: {}", m_namespace);

             trySync();

             Cat.logEvent("Apollo.Client.Version", Apollo.VERSION);

           }

         }, m_configUtil.getRefreshInterval(), m_configUtil.getRefreshInterval(),

         m_configUtil.getRefreshIntervalTimeUnit());

   }

   @Override

   protected synchronized void sync() {

     Transaction transaction = Cat.newTransaction("Apollo.ConfigService", "syncRemoteConfig");

     try {

       ApolloConfig previous = m_configCache.get();

       ApolloConfig current = loadApolloConfig();

       //reference equals means HTTP 304

       if (previous != current) {

         logger.debug("Remote Config refreshed!");

         m_configCache.set(current);

         this.fireRepositoryChange(m_namespace, this.getConfig());

       }

       transaction.setStatus(Message.SUCCESS);

     } catch (Throwable ex) {

       transaction.setStatus(ex);

       throw ex;

     } finally {

       transaction.complete();

     }

   }

   private Properties transformApolloConfigToProperties(ApolloConfig apolloConfig) {

     Properties result = new Properties();

     result.putAll(apolloConfig.getConfigurations());

     return result;

   }

   private ApolloConfig loadApolloConfig() {

     if (!m_loadConfigRateLimiter.tryAcquire(5, TimeUnit.SECONDS)) {

       //wait at most 5 seconds

       try {

         TimeUnit.SECONDS.sleep(5);

       } catch (InterruptedException e) {

       }

     }

     String appId = m_configUtil.getAppId();

     String cluster = m_configUtil.getCluster();

     String dataCenter = m_configUtil.getDataCenter();

     Cat.logEvent("Apollo.Client.ConfigInfo", STRING_JOINER.join(appId, cluster, m_namespace));

     int maxRetries = 2;

     Throwable exception = null;

     List<ServiceDTO> configServices = getConfigServices();

     for (int i = 0; i < maxRetries; i++) {

       List<ServiceDTO> randomConfigServices = Lists.newLinkedList(configServices);

       Collections.shuffle(randomConfigServices);

       //Access the server which notifies the client first

       if (m_longPollServiceDto.get() != null) {

         randomConfigServices.add(0, m_longPollServiceDto.getAndSet(null));

       }

       for (ServiceDTO configService : randomConfigServices) {

         String url =

             assembleQueryConfigUrl(configService.getHomepageUrl(), appId, cluster, m_namespace,

                 dataCenter, m_configCache.get());

         logger.debug("Loading config from {}", url);

         HttpRequest request = new HttpRequest(url);

         Transaction transaction = Cat.newTransaction("Apollo.ConfigService", "queryConfig");

         transaction.addData("Url", url);

         try {

           HttpResponse<ApolloConfig> response = m_httpUtil.doGet(request, ApolloConfig.class);

           transaction.addData("StatusCode", response.getStatusCode());

           transaction.setStatus(Message.SUCCESS);

           if (response.getStatusCode() == 304) {

             logger.debug("Config server responds with 304 HTTP status code.");

             return m_configCache.get();

           }

           ApolloConfig result = response.getBody();

           Cat.logEvent("Apollo.Client.ConfigLoaded." + result.getNamespaceName(),

               result.getReleaseKey());

           logger.debug("Loaded config for {}: {}", m_namespace, result);

           return result;

         } catch (Throwable ex) {

           Cat.logError(ex);

           transaction.setStatus(ex);

           exception = ex;

         } finally {

           transaction.complete();

         }

       }

       try {

         TimeUnit.SECONDS.sleep(1);

       } catch (InterruptedException ex) {

         //ignore

       }

     }

     String message = String.format(

         "Load Apollo Config failed - appId: %s, cluster: %s, namespace: %s, services: %s",

         appId, cluster, m_namespace, configServices);

     throw new ApolloConfigException(message, exception);

   }

   String assembleQueryConfigUrl(String uri, String appId, String cluster, String namespace,

                                         String dataCenter, ApolloConfig previousConfig) {

     String path = "configs/%s/%s/%s";

     List<String> pathParams =

         Lists.newArrayList(pathEscaper.escape(appId), pathEscaper.escape(cluster),

             pathEscaper.escape(namespace));

     Map<String, String> queryParams = Maps.newHashMap();

     if (previousConfig != null) {

       queryParams.put("releaseKey", queryParamEscaper.escape(previousConfig.getReleaseKey()));

     }

     if (!Strings.isNullOrEmpty(dataCenter)) {

       queryParams.put("dataCenter", queryParamEscaper.escape(dataCenter));

     }

     String localIp = m_configUtil.getLocalIp();

     if (!Strings.isNullOrEmpty(localIp)) {

       queryParams.put("ip", queryParamEscaper.escape(localIp));

     }

     String pathExpanded = String.format(path, pathParams.toArray());

     if (!queryParams.isEmpty()) {

       pathExpanded += "?" + MAP_JOINER.join(queryParams);

     }

     if (!uri.endsWith("/")) {

       uri += "/";

     }

     return uri + pathExpanded;

   }

   private void scheduleLongPollingRefresh() {

     try {

       final String appId = m_configUtil.getAppId();

       final String cluster = m_configUtil.getCluster();

       final String dataCenter = m_configUtil.getDataCenter();

       m_longPollingService.submit(new Runnable() {

         @Override

         public void run() {

           doLongPollingRefresh(appId, cluster, dataCenter);

         }

       });

     } catch (Throwable ex) {

       ApolloConfigException exception =

           new ApolloConfigException("Schedule long polling refresh failed", ex);

       Cat.logError(exception);

       logger.warn(ExceptionUtil.getDetailMessage(exception));

     }

   }

   private void doLongPollingRefresh(String appId, String cluster, String dataCenter) {

     final Random random = new Random();

     ServiceDTO lastServiceDto = null;

     while (!m_longPollingStopped.get() && !Thread.currentThread().isInterrupted()) {

       if (!m_longPollRateLimiter.tryAcquire(5, TimeUnit.SECONDS)) {

         //wait at most 5 seconds

         try {

           TimeUnit.SECONDS.sleep(5);

         } catch (InterruptedException e) {

         }

       }

       Transaction transaction = Cat.newTransaction("Apollo.ConfigService", "pollNotification");

       try {

         if (lastServiceDto == null) {

           List<ServiceDTO> configServices = getConfigServices();

           lastServiceDto = configServices.get(random.nextInt(configServices.size()));

         }

         String url =

             assembleLongPollRefreshUrl(lastServiceDto.getHomepageUrl(), appId, cluster,

                 m_namespace, dataCenter, m_longPollResult.get());

         logger.debug("Long polling from {}", url);

         HttpRequest request = new HttpRequest(url);

         //longer timeout for read - 1 minute

         request.setReadTimeout(60000);

         transaction.addData("Url", url);

         HttpResponse<ApolloConfigNotification> response =

             m_httpUtil.doGet(request, ApolloConfigNotification.class);

         logger.debug("Long polling response: {}, url: {}", response.getStatusCode(), url);

         if (response.getStatusCode() == 200) {

           m_longPollServiceDto.set(lastServiceDto);

           if (response.getBody() != null) {

             m_longPollResult.set(response.getBody());

             transaction.addData("Result", response.getBody().toString());

           }

           m_executorService.submit(new Runnable() {

             @Override

             public void run() {

               trySync();

             }

           });

         }

         m_longPollFailSchedulePolicyInSecond.success();

         transaction.addData("StatusCode", response.getStatusCode());

         transaction.setStatus(Message.SUCCESS);

       } catch (Throwable ex) {

         lastServiceDto = null;

         Cat.logError(ex);

         transaction.setStatus(ex);

         long sleepTimeInSecond = m_longPollFailSchedulePolicyInSecond.fail();

         logger.warn(

             "Long polling failed, will retry in {} seconds. appId: {}, cluster: {}, namespace: {}, reason: {}",

             sleepTimeInSecond, appId, cluster, m_namespace, ExceptionUtil.getDetailMessage(ex));

         try {

           TimeUnit.SECONDS.sleep(sleepTimeInSecond);

         } catch (InterruptedException ie) {

           //ignore

         }

       } finally {

         transaction.complete();

       }

     }

   }

   String assembleLongPollRefreshUrl(String uri, String appId, String cluster,

                                             String namespace, String dataCenter,

                                             ApolloConfigNotification previousResult) {

     Map<String, String> queryParams = Maps.newHashMap();

     queryParams.put("appId", queryParamEscaper.escape(appId));

     queryParams.put("cluster", queryParamEscaper.escape(cluster));

     if (!Strings.isNullOrEmpty(namespace)) {

       queryParams.put("namespace", queryParamEscaper.escape(namespace));

     }

     if (!Strings.isNullOrEmpty(dataCenter)) {

       queryParams.put("dataCenter", queryParamEscaper.escape(dataCenter));

     }

     String localIp = m_configUtil.getLocalIp();

     if (!Strings.isNullOrEmpty(localIp)) {

       queryParams.put("ip", queryParamEscaper.escape(localIp));

     }

     if (previousResult != null) {

       //number doesn't need encode

       queryParams.put("notificationId", String.valueOf(previousResult.getNotificationId()));

     }

     String params = MAP_JOINER.join(queryParams);

     if (!uri.endsWith("/")) {

       uri += "/";

     }

     return uri + "notifications?" + params;

   }

   void stopLongPollingRefresh() {

     this.m_longPollingStopped.compareAndSet(false, true);

   }

   private List<ServiceDTO> getConfigServices() {

     List<ServiceDTO> services = m_serviceLocator.getConfigServices();

     if (services.size() == 0) {

       throw new ApolloConfigException("No available config service");

     }

     return services;

   }

 }