##### Coverage Report - com.ctrip.apollo.internals.RemoteConfigRepository

Classes in this File | Line Coverage | Branch Coverage | Complexity
RemoteConfigRepository |
94% | 156/165
78% | 33/42
| 3.0625;3.062
RemoteConfigRepository$1 |
100% | 8/8
100% | 2/2
| 3.0625;3.062
RemoteConfigRepository$2 |
100% | 3/3
N/A | N/A
| 3.0625;3.062
RemoteConfigRepository$3 |
100% | 3/3
N/A | N/A
| 3.0625;3.062

1 |  |
     package com.ctrip.apollo.internals;
2 |  |
3 |  |
     import com.google.common.base.Joiner;
4 |  |
     import com.google.common.base.Strings;
5 |  |
     import com.google.common.collect.Lists;
6 |  |
     import com.google.common.collect.Maps;
7 |  |
     import com.google.common.escape.Escaper;
8 |  |
     import com.google.common.net.UrlEscapers;
9 |  |
10 |  |
     import com.ctrip.apollo.core.ConfigConsts;
11 |  |
     import com.ctrip.apollo.core.dto.ApolloConfig;
12 |  |
     import com.ctrip.apollo.core.dto.ApolloConfigNotification;
13 |  |
     import com.ctrip.apollo.core.dto.ServiceDTO;
14 |  |
     import com.ctrip.apollo.core.schedule.ExponentialSchedulePolicy;
15 |  |
     import com.ctrip.apollo.core.schedule.SchedulePolicy;
16 |  |
     import com.ctrip.apollo.core.utils.ApolloThreadFactory;
17 |  |
     import com.ctrip.apollo.util.ConfigUtil;
18 |  |
     import com.ctrip.apollo.util.ExceptionUtil;
19 |  |
     import com.ctrip.apollo.util.http.HttpRequest;
20 |  |
     import com.ctrip.apollo.util.http.HttpResponse;
21 |  |
     import com.ctrip.apollo.util.http.HttpUtil;
22 |  |
     import com.dianping.cat.Cat;
23 |  |
     import com.dianping.cat.message.Message;
24 |  |
     import com.dianping.cat.message.Transaction;
25 |  |
26 |  |
     import org.codehaus.plexus.PlexusContainer;
27 |  |
     import org.codehaus.plexus.component.repository.exception.ComponentLookupException;
28 |  |
     import org.slf4j.Logger;
29 |  |
     import org.slf4j.LoggerFactory;
30 |  |
     import org.unidal.lookup.ContainerLoader;
31 |  |
32 |  |
     import java.util.Collections;
33 |  |
     import java.util.List;
34 |  |
     import java.util.Map;
35 |  |
     import java.util.Properties;
36 |  |
     import java.util.Random;
37 |  |
     import java.util.concurrent.ExecutorService;
38 |  |
     import java.util.concurrent.Executors;
39 |  |
     import java.util.concurrent.ScheduledExecutorService;
40 |  |
     import java.util.concurrent.TimeUnit;
41 |  |
     import java.util.concurrent.atomic.AtomicBoolean;
42 |  |
     import java.util.concurrent.atomic.AtomicReference;
43 |  |
44 |  |
     /**
45 |  |
      * @author Jason Song(song_s@ctrip.com)
46 |  |
      */
47 | 42 |
     public class RemoteConfigRepository extends AbstractConfigRepository {
48 | 1 |
       private static final Logger logger = LoggerFactory.getLogger(RemoteConfigRepository.class);
49 | 1 |
       private static final Joiner STRING_JOINER = Joiner.on(ConfigConsts.CLUSTER_NAMESPACE_SEPARATOR);
50 | 1 |
       private static final Joiner.MapJoiner MAP_JOINER = Joiner.on("&").withKeyValueSeparator("=");
51 |  |
       private PlexusContainer m_container;
52 |  |
       private final ConfigServiceLocator m_serviceLocator;
53 |  |
       private final HttpUtil m_httpUtil;
54 |  |
       private final ConfigUtil m_configUtil;
55 |  |
       private volatile AtomicReference<ApolloConfig> m_configCache;
56 |  |
       private final String m_namespace;
57 |  |
       private final ScheduledExecutorService m_executorService;
58 |  |
       private final AtomicBoolean m_longPollingStopped;
59 |  |
       private SchedulePolicy m_longPollSchedulePolicy;
60 |  |
       private AtomicReference<ServiceDTO> m_longPollServiceDto;
61 |  |
62 |  |
       /**
63 |  |
        * Constructor.
64 |  |
        *
65 |  |
        * @param namespace the namespace
66 |  |
        */
67 | 12 |
       public RemoteConfigRepository(String namespace) {
68 | 12 |
         m_namespace = namespace;
69 | 12 |
         m_configCache = new AtomicReference<>();
70 | 12 |
         m_container = ContainerLoader.getDefaultContainer();
71 |  |
         try {
72 | 12 |
           m_configUtil = m_container.lookup(ConfigUtil.class);
73 | 12 |
           m_httpUtil = m_container.lookup(HttpUtil.class);
74 | 12 |
           m_serviceLocator = m_container.lookup(ConfigServiceLocator.class);
75 | 0 |
         } catch (ComponentLookupException ex) {
76 | 0 |
           Cat.logError(ex);
77 | 0 |
           throw new IllegalStateException("Unable to load component!", ex);
78 | 12 |
         }
79 | 12 |
         m_longPollSchedulePolicy = new ExponentialSchedulePolicy(1, 120);
80 | 12 |
         m_longPollingStopped = new AtomicBoolean(false);
81 | 24 |
         m_executorService = Executors.newScheduledThreadPool(1,
82 | 12 |
             ApolloThreadFactory.create("RemoteConfigRepository", true));
83 | 12 |
         m_longPollServiceDto = new AtomicReference<>();
84 | 12 |
         this.trySync();
85 | 12 |
         this.schedulePeriodicRefresh();
86 | 12 |
         this.scheduleLongPollingRefresh();
87 | 12 |
       }
88 |  |
89 |  |
       @Override
90 |  |
       public Properties getConfig() {
91 | 32 |
         if (m_configCache.get() == null) {
92 | 4 |
           this.sync();
93 |  |
         }
94 | 29 |
         return transformApolloConfigToProperties(m_configCache.get());
95 |  |
       }
96 |  |
97 |  |
       @Override
98 |  |
       public void setFallback(ConfigRepository fallbackConfigRepository) {
99 |  |
         //remote config doesn't need fallback
100 | 0 |
       }
101 |  |
102 |  |
       private void schedulePeriodicRefresh() {
103 | 24 |
         logger.debug("Schedule periodic refresh with interval: {} {}",
104 | 12 |
             m_configUtil.getRefreshInterval(), m_configUtil.getRefreshTimeUnit());
105 | 24 |
         this.m_executorService.scheduleAtFixedRate(
106 | 12 |
             new Runnable() {
107 |  |
               @Override
108 |  |
               public void run() {
109 | 15 |
                 logger.debug("refresh config for namespace: {}", m_namespace);
110 | 15 |
                 Transaction transaction = Cat.newTransaction("Apollo.ConfigService", "periodicRefresh");
111 | 15 |
                 boolean syncSuccess = trySync();
112 | 14 |
                 String status = syncSuccess ? Message.SUCCESS : "-1";
113 | 14 |
                 transaction.setStatus(status);
114 | 14 |
                 transaction.complete();
115 | 14 |
               }
116 | 12 |
             }, m_configUtil.getRefreshInterval(), m_configUtil.getRefreshInterval(),
117 | 12 |
             m_configUtil.getRefreshTimeUnit());
118 | 12 |
       }
119 |  |
120 |  |
       @Override
121 |  |
       protected synchronized void sync() {
122 | 34 |
         ApolloConfig previous = m_configCache.get();
123 | 34 |
         ApolloConfig current = loadApolloConfig();
124 |  |
125 |  |
         //HTTP 304, nothing changed
126 | 23 |
         if (previous == current) {
127 | 2 |
           return;
128 |  |
         }
129 |  |
130 | 21 |
         logger.debug("Remote Config refreshed!");
131 |  |
132 | 21 |
         m_configCache.set(current);
133 |  |
134 | 21 |
         this.fireRepositoryChange(m_namespace, this.getConfig());
135 | 21 |
       }
136 |  |
137 |  |
       private Properties transformApolloConfigToProperties(ApolloConfig apolloConfig) {
138 | 29 |
         Properties result = new Properties();
139 | 29 |
         result.putAll(apolloConfig.getConfigurations());
140 | 28 |
         return result;
141 |  |
       }
142 |  |
143 |  |
       private ApolloConfig loadApolloConfig() {
144 | 34 |
         String appId = m_configUtil.getAppId();
145 | 34 |
         String cluster = m_configUtil.getCluster();
146 | 34 |
         String dataCenter = m_configUtil.getDataCenter();
147 | 34 |
         Cat.logEvent("Apollo.Client.ConfigInfo", STRING_JOINER.join(appId, cluster, m_namespace));
148 | 34 |
         int maxRetries = 2;
149 | 34 |
         Throwable exception = null;
150 |  |
151 | 34 |
         List<ServiceDTO> configServices = getConfigServices();
152 | 56 |
         for (int i = 0; i < maxRetries; i++) {
153 | 46 |
           List<ServiceDTO> randomConfigServices = Lists.newLinkedList(configServices);
154 | 46 |
           Collections.shuffle(randomConfigServices);
155 |  |
           //Access the server which notifies the client first
156 | 46 |
           if (m_longPollServiceDto.get() != null) {
157 | 2 |
             randomConfigServices.add(0, m_longPollServiceDto.getAndSet(null));
158 |  |
           }
159 |  |
160 | 46 |
           for (ServiceDTO configService : randomConfigServices) {
161 | 46 |
             String url =
162 | 92 |
                 assembleQueryConfigUrl(configService.getHomepageUrl(), appId, cluster, m_namespace,
163 | 46 |
                     dataCenter, m_configCache.get());
164 |  |
165 | 46 |
             logger.debug("Loading config from {}", url);
166 | 46 |
             HttpRequest request = new HttpRequest(url);
167 |  |
168 | 46 |
             Transaction transaction = Cat.newTransaction("Apollo.ConfigService", "queryConfig");
169 | 46 |
             transaction.addData("Url", url);
170 |  |
             try {
171 |  |
172 | 46 |
               HttpResponse<ApolloConfig> response = m_httpUtil.doGet(request, ApolloConfig.class);
173 |  |
174 | 23 |
               transaction.addData("StatusCode", response.getStatusCode());
175 | 23 |
               transaction.setStatus(Message.SUCCESS);
176 |  |
177 | 23 |
               if (response.getStatusCode() == 304) {
178 | 0 |
                 logger.debug("Config server responds with 304 HTTP status code.");
179 | 0 |
                 return m_configCache.get();
180 |  |
               }
181 | 23 |
               logger.debug("Loaded config: {}", response.getBody());
182 |  |
183 | 23 |
               return response.getBody();
184 | 23 |
             } catch (Throwable ex) {
185 | 23 |
               Cat.logError(ex);
186 | 23 |
               transaction.setStatus(ex);
187 | 23 |
               exception = ex;
188 |  |
             } finally {
189 | 46 |
               transaction.complete();
190 | 23 |
             }
191 |  |
192 | 23 |
           }
193 |  |
194 |  |
           try {
195 | 23 |
             TimeUnit.SECONDS.sleep(1);
196 | 0 |
           } catch (InterruptedException ex) {
197 |  |
             //ignore
198 | 22 |
           }
199 |  |
         }
200 | 10 |
         String message = String.format(
201 |  |
             "Load Apollo Config failed - appId: %s, cluster: %s, namespace: %s, services: %s",
202 |  |
             appId, cluster, m_namespace, configServices);
203 | 10 |
         throw new RuntimeException(message, exception);
204 |  |
       }
205 |  |
206 |  |
       private String assembleQueryConfigUrl(String uri, String appId, String cluster, String namespace,
207 |  |
                                             String dataCenter, ApolloConfig previousConfig) {
208 | 46 |
         Escaper escaper = UrlEscapers.urlPathSegmentEscaper();
209 | 46 |
         String path = "configs/%s/%s";
210 | 46 |
         List<String> pathParams = Lists.newArrayList(escaper.escape(appId), escaper.escape(cluster));
211 | 46 |
         Map<String, String> queryParams = Maps.newHashMap();
212 |  |
213 | 46 |
         if (!Strings.isNullOrEmpty(namespace)) {
214 | 46 |
           path = path + "/%s";
215 | 46 |
           pathParams.add(escaper.escape(namespace));
216 |  |
         }
217 |  |
218 | 46 |
         if (previousConfig != null) {
219 | 24 |
           queryParams.put("releaseKey", escaper.escape(String.valueOf(previousConfig.getReleaseKey())));
220 |  |
         }
221 |  |
222 | 46 |
         if (!Strings.isNullOrEmpty(dataCenter)) {
223 | 39 |
           queryParams.put("dataCenter", escaper.escape(dataCenter));
224 |  |
         }
225 |  |
226 | 46 |
         String pathExpanded = String.format(path, pathParams.toArray());
227 |  |
228 | 46 |
         if (!queryParams.isEmpty()) {
229 | 41 |
           pathExpanded += "?" + MAP_JOINER.join(queryParams);
230 |  |
         }
231 | 46 |
         if (!uri.endsWith("/")) {
232 | 46 |
           uri += "/";
233 |  |
         }
234 | 46 |
         return uri + pathExpanded;
235 |  |
       }
236 |  |
237 |  |
       private void scheduleLongPollingRefresh() {
238 | 12 |
         final String appId = m_configUtil.getAppId();
239 | 12 |
         final String cluster = m_configUtil.getCluster();
240 | 12 |
         final String dataCenter = m_configUtil.getDataCenter();
241 | 12 |
         final ExecutorService longPollingService =
242 | 12 |
             Executors.newFixedThreadPool(2,
243 | 12 |
                 ApolloThreadFactory.create("RemoteConfigRepository-LongPolling", true));
244 | 12 |
         longPollingService.submit(new Runnable() {
245 |  |
           @Override
246 |  |
           public void run() {
247 | 12 |
             doLongPollingRefresh(appId, cluster, dataCenter, longPollingService);
248 | 4 |
           }
249 |  |
         });
250 | 12 |
       }
251 |  |
252 |  |
       private void doLongPollingRefresh(String appId, String cluster, String dataCenter,
253 |  |
                                         ExecutorService longPollingService) {
254 | 12 |
         final Random random = new Random();
255 | 12 |
         ServiceDTO lastServiceDto = null;
256 | 12 |
         Transaction transaction = null;
257 | 36 |
         while (!m_longPollingStopped.get() && !Thread.currentThread().isInterrupted()) {
258 |  |
           try {
259 | 32 |
             if (lastServiceDto == null) {
260 | 30 |
               List<ServiceDTO> configServices = getConfigServices();
261 | 30 |
               lastServiceDto = configServices.get(random.nextInt(configServices.size()));
262 |  |
             }
263 |  |
264 | 32 |
             String url =
265 | 32 |
                 assembleLongPollRefreshUrl(lastServiceDto.getHomepageUrl(), appId, cluster,
266 |  |
                     m_namespace, dataCenter);
267 |  |
268 | 32 |
             logger.debug("Long polling from {}", url);
269 | 32 |
             HttpRequest request = new HttpRequest(url);
270 |  |
             //no timeout for read
271 | 32 |
             request.setReadTimeout(0);
272 |  |
273 | 32 |
             transaction = Cat.newTransaction("Apollo.ConfigService", "pollNotification");
274 | 32 |
             transaction.addData("Url", url);
275 |  |
276 | 32 |
             HttpResponse<ApolloConfigNotification> response =
277 | 32 |
                 m_httpUtil.doGet(request, ApolloConfigNotification.class);
278 |  |
279 | 3 |
             logger.debug("Long polling response: {}, url: {}", response.getStatusCode(), url);
280 | 3 |
             if (response.getStatusCode() == 200) {
281 | 2 |
               m_longPollServiceDto.set(lastServiceDto);
282 | 2 |
               longPollingService.submit(new Runnable() {
283 |  |
                 @Override
284 |  |
                 public void run() {
285 | 2 |
                   trySync();
286 | 2 |
                 }
287 |  |
               });
288 |  |
             }
289 | 3 |
             m_longPollSchedulePolicy.success();
290 | 3 |
             transaction.addData("StatusCode", response.getStatusCode());
291 | 3 |
             transaction.setStatus(Message.SUCCESS);
292 | 29 |
           } catch (Throwable ex) {
293 | 29 |
             lastServiceDto = null;
294 | 29 |
             Cat.logError(ex);
295 | 29 |
             if (transaction != null) {
296 | 29 |
               transaction.setStatus(ex);
297 |  |
             }
298 | 29 |
             long sleepTime = m_longPollSchedulePolicy.fail();
299 | 58 |
             logger.warn(
300 |  |
                 "Long polling failed, will retry in {} seconds. appId: {}, cluster: {}, namespace: {}, reason: {}",
301 | 29 |
                 sleepTime, appId, cluster, m_namespace, ExceptionUtil.getDetailMessage(ex));
302 |  |
             try {
303 | 29 |
               TimeUnit.SECONDS.sleep(sleepTime);
304 | 0 |
             } catch (InterruptedException ie) {
305 |  |
               //ignore
306 | 21 |
             }
307 |  |
           } finally {
308 | 24 |
             if (transaction != null) {
309 | 24 |
               transaction.complete();
310 |  |
             }
311 |  |
           }
312 |  |
         }
313 | 4 |
       }
314 |  |
315 |  |
       private String assembleLongPollRefreshUrl(String uri, String appId, String cluster,
316 |  |
                                                 String namespace, String dataCenter) {
317 | 32 |
         Escaper escaper = UrlEscapers.urlPathSegmentEscaper();
318 | 32 |
         Map<String, String> queryParams = Maps.newHashMap();
319 | 32 |
         queryParams.put("appId", escaper.escape(appId));
320 | 32 |
         queryParams.put("cluster", escaper.escape(cluster));
321 |  |
322 | 32 |
         if (!Strings.isNullOrEmpty(namespace)) {
323 | 32 |
           queryParams.put("namespace", escaper.escape(namespace));
324 |  |
         }
325 | 32 |
         if (!Strings.isNullOrEmpty(dataCenter)) {
326 | 30 |
           queryParams.put("dataCenter", escaper.escape(dataCenter));
327 |  |
         }
328 |  |
329 | 32 |
         String params = MAP_JOINER.join(queryParams);
330 | 32 |
         if (!uri.endsWith("/")) {
331 | 32 |
           uri += "/";
332 |  |
         }
333 |  |
334 | 32 |
         return uri + "notifications?" + params;
335 |  |
       }
336 |  |
337 |  |
       void stopLongPollingRefresh() {
338 | 4 |
         this.m_longPollingStopped.compareAndSet(false, true);
339 | 4 |
       }
340 |  |
341 |  |
       private List<ServiceDTO> getConfigServices() {
342 | 64 |
         List<ServiceDTO> services = m_serviceLocator.getConfigServices();
343 | 64 |
         if (services.size() == 0) {
344 | 0 |
           throw new RuntimeException("No available config service");
345 |  |
         }
346 |  |
347 | 64 |
         return services;
348 |  |
       }
349 |  |
     }