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

Classes in this File | Line Coverage | Branch Coverage | Complexity
NotificationController |
94% | 64/68
76% | 23/30
| 0.0;0

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