##### Coverage Report - com.ctrip.apollo.configservice.controller.NotificationController

Classes in this File | Line Coverage | Branch Coverage | Complexity
NotificationController |
93% | 56/60
77% | 17/22
| 0.0;0

1 |  |
     package com.ctrip.apollo.configservice.controller;
2 |  |
3 |  |
     import com.google.common.base.Joiner;
4 |  |
     import com.google.common.base.Splitter;
5 |  |
     import com.google.common.base.Strings;
6 |  |
     import com.google.common.collect.HashMultimap;
7 |  |
     import com.google.common.collect.Lists;
8 |  |
     import com.google.common.collect.Multimap;
9 |  |
     import com.google.common.collect.Multimaps;
10 |  |
11 |  |
     import com.ctrip.apollo.biz.entity.AppNamespace;
12 |  |
     import com.ctrip.apollo.biz.message.MessageListener;
13 |  |
     import com.ctrip.apollo.biz.message.Topics;
14 |  |
     import com.ctrip.apollo.biz.service.AppNamespaceService;
15 |  |
     import com.ctrip.apollo.biz.utils.EntityManagerUtil;
16 |  |
     import com.ctrip.apollo.core.ConfigConsts;
17 |  |
     import com.ctrip.apollo.core.dto.ApolloConfigNotification;
18 |  |
     import com.dianping.cat.Cat;
19 |  |
20 |  |
     import org.slf4j.Logger;
21 |  |
     import org.slf4j.LoggerFactory;
22 |  |
     import org.springframework.beans.factory.annotation.Autowired;
23 |  |
     import org.springframework.http.HttpStatus;
24 |  |
     import org.springframework.http.ResponseEntity;
25 |  |
     import org.springframework.web.bind.annotation.RequestMapping;
26 |  |
     import org.springframework.web.bind.annotation.RequestMethod;
27 |  |
     import org.springframework.web.bind.annotation.RequestParam;
28 |  |
     import org.springframework.web.bind.annotation.RestController;
29 |  |
     import org.springframework.web.context.request.async.DeferredResult;
30 |  |
31 |  |
     import java.util.List;
32 |  |
     import java.util.Objects;
33 |  |
34 |  |
     /**
35 |  |
      * @author Jason Song(song_s@ctrip.com)
36 |  |
      */
37 |  |
     @RestController
38 |  |
     @RequestMapping("/notifications")
39 | 5 |
     public class NotificationController implements MessageListener {
40 | 1 |
       private static final Logger logger = LoggerFactory.getLogger(NotificationController.class);
41 |  |
       private static final long TIMEOUT = 360 * 60 * 1000;//6 hours
42 | 10 |
       private final Multimap<String, DeferredResult<ResponseEntity<ApolloConfigNotification>>>
43 | 5 |
           deferredResults = Multimaps.synchronizedSetMultimap(HashMultimap.create());
44 |  |
       private static final ResponseEntity<ApolloConfigNotification>
45 | 1 |
           NOT_MODIFIED_RESPONSE = new ResponseEntity<>(HttpStatus.NOT_MODIFIED);
46 | 1 |
       private static final Joiner STRING_JOINER = Joiner.on(ConfigConsts.CLUSTER_NAMESPACE_SEPARATOR);
47 | 2 |
       private static final Splitter STRING_SPLITTER =
48 | 1 |
           Splitter.on(ConfigConsts.CLUSTER_NAMESPACE_SEPARATOR).omitEmptyStrings();
49 |  |
50 |  |
       @Autowired
51 |  |
       private AppNamespaceService appNamespaceService;
52 |  |
53 |  |
       @Autowired
54 |  |
       private EntityManagerUtil entityManagerUtil;
55 |  |
56 |  |
       @RequestMapping(method = RequestMethod.GET)
57 |  |
       public DeferredResult<ResponseEntity<ApolloConfigNotification>> pollNotification(
58 |  |
           @RequestParam(value = "appId") String appId,
59 |  |
           @RequestParam(value = "cluster") String cluster,
60 |  |
           @RequestParam(value = "namespace", defaultValue = ConfigConsts.NAMESPACE_DEFAULT) String namespace,
61 |  |
           @RequestParam(value = "dataCenter", required = false) String dataCenter) {
62 | 7 |
         List<String> watchedKeys = Lists.newArrayList(assembleKey(appId, cluster, namespace));
63 |  |
64 |  |
         //Listen on more namespaces, since it's not the default namespace
65 | 7 |
         if (!Objects.equals(ConfigConsts.NAMESPACE_DEFAULT, namespace)) {
66 | 4 |
           watchedKeys.addAll(this.findPublicConfigWatchKey(appId, namespace, dataCenter));
67 |  |
         }
68 |  |
69 | 7 |
         DeferredResult<ResponseEntity<ApolloConfigNotification>> deferredResult =
70 | 7 |
             new DeferredResult<>(TIMEOUT, NOT_MODIFIED_RESPONSE);
71 |  |
72 |  |
         //register all keys
73 | 7 |
         for (String key : watchedKeys) {
74 | 14 |
           this.deferredResults.put(key, deferredResult);
75 | 14 |
         }
76 |  |
77 | 7 |
         deferredResult.onTimeout(() -> logWatchedKeysToCat(watchedKeys, "Apollo.LongPoll.TimeOutKeys"));
78 |  |
79 | 7 |
         deferredResult.onCompletion(() -> {
80 |  |
           //unregister all keys
81 | 3 |
           for (String key : watchedKeys) {
82 | 6 |
             deferredResults.remove(key, deferredResult);
83 | 6 |
           }
84 | 3 |
           logWatchedKeysToCat(watchedKeys, "Apollo.LongPoll.CompletedKeys");
85 | 3 |
         });
86 |  |
87 | 7 |
         logWatchedKeysToCat(watchedKeys, "Apollo.LongPoll.RegisteredKeys");
88 | 7 |
         logger.info("Listening {} from appId: {}, cluster: {}, namespace: {}, datacenter: {}",
89 |  |
             watchedKeys, appId, cluster, namespace, dataCenter);
90 | 7 |
         return deferredResult;
91 |  |
       }
92 |  |
93 |  |
       private String assembleKey(String appId, String cluster, String namespace) {
94 | 14 |
         return STRING_JOINER.join(appId, cluster, namespace);
95 |  |
       }
96 |  |
97 |  |
       private List<String> findPublicConfigWatchKey(String applicationId, String namespace,
98 |  |
                                                     String dataCenter) {
99 | 4 |
         List<String> publicWatchedKeys = Lists.newArrayList();
100 | 4 |
         AppNamespace appNamespace = appNamespaceService.findByNamespaceName(namespace);
101 |  |
         /**
102 |  |
          * Manually close the entity manager.
103 |  |
          * Since for async request, Spring won't do so until the request is finished,
104 |  |
          * which is unacceptable since we are doing long polling - means the db connection would be hold
105 |  |
          * for a very long time
106 |  |
          */
107 | 4 |
         entityManagerUtil.closeEntityManager();
108 |  |
109 |  |
         //check whether the namespace's appId equals to current one
110 | 4 |
         if (Objects.isNull(appNamespace) || Objects.equals(applicationId, appNamespace.getAppId())) {
111 | 0 |
           return publicWatchedKeys;
112 |  |
         }
113 |  |
114 | 4 |
         String publicConfigAppId = appNamespace.getAppId();
115 |  |
116 |  |
         //watch data center config change
117 | 4 |
         if (!Objects.isNull(dataCenter)) {
118 | 3 |
           publicWatchedKeys.add(assembleKey(publicConfigAppId, dataCenter, namespace));
119 |  |
         }
120 |  |
121 |  |
         //watch default cluster config change
122 | 4 |
         publicWatchedKeys
123 | 4 |
             .add(assembleKey(publicConfigAppId, ConfigConsts.CLUSTER_NAME_DEFAULT, namespace));
124 |  |
125 | 4 |
         return publicWatchedKeys;
126 |  |
       }
127 |  |
128 |  |
       @Override
129 |  |
       public void handleMessage(String message, String channel) {
130 | 8 |
         logger.info("message received - channel: {}, message: {}", channel, message);
131 | 8 |
         Cat.logEvent("Apollo.LongPoll.Messages", message);
132 | 8 |
         if (!Topics.APOLLO_RELEASE_TOPIC.equals(channel) || Strings.isNullOrEmpty(message)) {
133 | 0 |
           return;
134 |  |
         }
135 | 8 |
         List<String> keys = STRING_SPLITTER.splitToList(message);
136 |  |
         //message should be appId|cluster|namespace
137 | 8 |
         if (keys.size() != 3) {
138 | 0 |
           logger.error("message format invalid - {}", message);
139 | 0 |
           return;
140 |  |
         }
141 |  |
142 | 8 |
         ResponseEntity<ApolloConfigNotification> notification =
143 |  |
             new ResponseEntity<>(
144 | 8 |
                 new ApolloConfigNotification(keys.get(2)), HttpStatus.OK);
145 |  |
146 |  |
         //create a new list to avoid ConcurrentModificationException
147 | 8 |
         List<DeferredResult<ResponseEntity<ApolloConfigNotification>>> results =
148 | 8 |
             Lists.newArrayList(deferredResults.get(message));
149 | 8 |
         logger.info("Notify {} clients for key {}", results.size(), message);
150 |  |
151 | 8 |
         for (DeferredResult<ResponseEntity<ApolloConfigNotification>> result : results) {
152 | 5 |
           result.setResult(notification);
153 | 5 |
         }
154 | 8 |
         logger.info("Notification completed");
155 | 8 |
       }
156 |  |
157 |  |
       private void logWatchedKeysToCat(List<String> watchedKeys, String eventName) {
158 | 10 |
         for (String watchedKey : watchedKeys) {
159 | 20 |
           Cat.logEvent(eventName, watchedKey);
160 | 20 |
         }
161 | 10 |
       }
162 |  |
     }
163 |  |