NotificationControllerV2 implements ReleaseMessageListener { @GetMapping public DeferredResult<ResponseEntity<List<ApolloConfigNotification>>> pollNotification( @RequestParam(value = "appId") String appId, @RequestParam(value = "cluster") String cluster, @RequestParam(value = "notifications") String notificationsAsString, @RequestParam(value = "dataCenter", required = false) String dataCenter, @RequestParam(value = "ip", required = false) String clientIp) { List<ApolloConfigNotification> notifications = null; try { notifications = gson.fromJson(notificationsAsString, notificationsTypeReference); } catch (Throwable ex) { Tracer.logError(ex); } if (CollectionUtils.isEmpty(notifications)) { throw new BadRequestException("Invalid format of notifications: " + notificationsAsString); } Map<String, ApolloConfigNotification> filteredNotifications = filterNotifications(appId, notifications); if (CollectionUtils.isEmpty(filteredNotifications)) { throw new BadRequestException("Invalid format of notifications: " + notificationsAsString); } DeferredResultWrapper deferredResultWrapper = new DeferredResultWrapper(bizConfig.longPollingTimeoutInMilli()); Set<String> namespaces = Sets.newHashSetWithExpectedSize(filteredNotifications.size()); Map<String, Long> clientSideNotifications = Maps.newHashMapWithExpectedSize(filteredNotifications.size()); for (Map.Entry<String, ApolloConfigNotification> notificationEntry : filteredNotifications.entrySet()) { String normalizedNamespace = notificationEntry.getKey(); ApolloConfigNotification notification = notificationEntry.getValue(); namespaces.add(normalizedNamespace); clientSideNotifications.put(normalizedNamespace, notification.getNotificationId()); if (!Objects.equals(notification.getNamespaceName(), normalizedNamespace)) { deferredResultWrapper.recordNamespaceNameNormalizedResult(notification.getNamespaceName(), normalizedNamespace); } } Multimap<String, String> watchedKeysMap = watchKeysUtil.assembleAllWatchKeys(appId, cluster, namespaces, dataCenter); Set<String> watchedKeys = Sets.newHashSet(watchedKeysMap.values()); deferredResultWrapper .onTimeout(() -> logWatchedKeys(watchedKeys, "Apollo.LongPoll.TimeOutKeys")); deferredResultWrapper.onCompletion(() -> { for (String key : watchedKeys) { deferredResults.remove(key, deferredResultWrapper); } logWatchedKeys(watchedKeys, "Apollo.LongPoll.CompletedKeys"); }); for (String key : watchedKeys) { this.deferredResults.put(key, deferredResultWrapper); } logWatchedKeys(watchedKeys, "Apollo.LongPoll.RegisteredKeys"); logger.debug("Listening {} from appId: {}, cluster: {}, namespace: {}, datacenter: {}", watchedKeys, appId, cluster, namespaces, dataCenter); List<ReleaseMessage> latestReleaseMessages = releaseMessageService.findLatestReleaseMessagesGroupByMessages(watchedKeys); entityManagerUtil.closeEntityManager(); List<ApolloConfigNotification> newNotifications = getApolloConfigNotifications(namespaces, clientSideNotifications, watchedKeysMap, latestReleaseMessages); if (!CollectionUtils.isEmpty(newNotifications)) { deferredResultWrapper.setResult(newNotifications); } return deferredResultWrapper.getResult(); } @Autowired  NotificationControllerV2(
      final WatchKeysUtil watchKeysUtil,
      final ReleaseMessageServiceWithCache releaseMessageService,
      final EntityManagerUtil entityManagerUtil,
      final NamespaceUtil namespaceUtil,
      final Gson gson,
      final BizConfig bizConfig); @GetMapping DeferredResult<ResponseEntity<List<ApolloConfigNotification>>> pollNotification(
      @RequestParam(value = "appId") String appId,
      @RequestParam(value = "cluster") String cluster,
      @RequestParam(value = "notifications") String notificationsAsString,
      @RequestParam(value = "dataCenter", required = false) String dataCenter,
      @RequestParam(value = "ip", required = false) String clientIp); @Override void handleMessage(ReleaseMessage message, String channel);  }
@Test public void testPollNotificationWithDefaultNamespace() throws Exception { String someWatchKey = "someKey"; String anotherWatchKey = "anotherKey"; Multimap<String, String> watchKeysMap = assembleMultiMap(defaultNamespace, Lists.newArrayList(someWatchKey, anotherWatchKey)); String notificationAsString = transformApolloConfigNotificationsToString(defaultNamespace, someNotificationId); when(watchKeysUtil .assembleAllWatchKeys(someAppId, someCluster, Sets.newHashSet(defaultNamespace), someDataCenter)).thenReturn( watchKeysMap); DeferredResult<ResponseEntity<List<ApolloConfigNotification>>> deferredResult = controller .pollNotification(someAppId, someCluster, notificationAsString, someDataCenter, someClientIp); assertEquals(watchKeysMap.size(), deferredResults.size()); assertWatchKeys(watchKeysMap, deferredResult); }
@Test public void testPollNotificationWithDefaultNamespaceAsFile() throws Exception { String namespace = String.format("%s.%s", defaultNamespace, "properties"); when(namespaceUtil.filterNamespaceName(namespace)).thenReturn(defaultNamespace); String someWatchKey = "someKey"; String anotherWatchKey = "anotherKey"; Multimap<String, String> watchKeysMap = assembleMultiMap(defaultNamespace, Lists.newArrayList(someWatchKey, anotherWatchKey)); String notificationAsString = transformApolloConfigNotificationsToString(namespace, someNotificationId); when(watchKeysUtil .assembleAllWatchKeys(someAppId, someCluster, Sets.newHashSet(defaultNamespace), someDataCenter)).thenReturn( watchKeysMap); DeferredResult<ResponseEntity<List<ApolloConfigNotification>>> deferredResult = controller .pollNotification(someAppId, someCluster, notificationAsString, someDataCenter, someClientIp); assertEquals(watchKeysMap.size(), deferredResults.size()); assertWatchKeys(watchKeysMap, deferredResult); }
@Test public void testPollNotificationWithMultipleNamespaces() throws Exception { String defaultNamespaceAsFile = defaultNamespace + ".properties"; String somePublicNamespaceAsFile = somePublicNamespace + ".xml"; when(namespaceUtil.filterNamespaceName(defaultNamespaceAsFile)).thenReturn(defaultNamespace); when(namespaceUtil.filterNamespaceName(somePublicNamespaceAsFile)).thenReturn(somePublicNamespaceAsFile); when(namespaceUtil.normalizeNamespace(someAppId, somePublicNamespaceAsFile)).thenReturn(somePublicNamespaceAsFile); String someWatchKey = "someKey"; String anotherWatchKey = "anotherKey"; String somePublicWatchKey = "somePublicWatchKey"; String somePublicFileWatchKey = "somePublicFileWatchKey"; Multimap<String, String> watchKeysMap = assembleMultiMap(defaultNamespace, Lists.newArrayList(someWatchKey, anotherWatchKey)); watchKeysMap .putAll(assembleMultiMap(somePublicNamespace, Lists.newArrayList(somePublicWatchKey))); watchKeysMap .putAll(assembleMultiMap(somePublicNamespaceAsFile, Lists.newArrayList(somePublicFileWatchKey))); String notificationAsString = transformApolloConfigNotificationsToString(defaultNamespaceAsFile, someNotificationId, somePublicNamespace, someNotificationId, somePublicNamespaceAsFile, someNotificationId); when(watchKeysUtil .assembleAllWatchKeys(someAppId, someCluster, Sets.newHashSet(defaultNamespace, somePublicNamespace, somePublicNamespaceAsFile), someDataCenter)).thenReturn( watchKeysMap); DeferredResult<ResponseEntity<List<ApolloConfigNotification>>> deferredResult = controller .pollNotification(someAppId, someCluster, notificationAsString, someDataCenter, someClientIp); assertEquals(watchKeysMap.size(), deferredResults.size()); assertWatchKeys(watchKeysMap, deferredResult); verify(watchKeysUtil, times(1)).assembleAllWatchKeys(someAppId, someCluster, Sets.newHashSet(defaultNamespace, somePublicNamespace, somePublicNamespaceAsFile), someDataCenter); }
@Test public void testPollNotificationWithMultipleNamespaceWithNotificationIdOutDated() throws Exception { String someWatchKey = "someKey"; String anotherWatchKey = Joiner.on(ConfigConsts.CLUSTER_NAMESPACE_SEPARATOR) .join(someAppId, someCluster, somePublicNamespace); String yetAnotherWatchKey = Joiner.on(ConfigConsts.CLUSTER_NAMESPACE_SEPARATOR) .join(someAppId, defaultCluster, somePublicNamespace); long notificationId = someNotificationId + 1; long yetAnotherNotificationId = someNotificationId; Multimap<String, String> watchKeysMap = assembleMultiMap(defaultNamespace, Lists.newArrayList(someWatchKey)); watchKeysMap .putAll(assembleMultiMap(somePublicNamespace, Lists.newArrayList(anotherWatchKey, yetAnotherWatchKey))); when(watchKeysUtil .assembleAllWatchKeys(someAppId, someCluster, Sets.newHashSet(defaultNamespace, somePublicNamespace), someDataCenter)).thenReturn( watchKeysMap); ReleaseMessage someReleaseMessage = mock(ReleaseMessage.class); when(someReleaseMessage.getId()).thenReturn(notificationId); when(someReleaseMessage.getMessage()).thenReturn(anotherWatchKey); ReleaseMessage yetAnotherReleaseMessage = mock(ReleaseMessage.class); when(yetAnotherReleaseMessage.getId()).thenReturn(yetAnotherNotificationId); when(yetAnotherReleaseMessage.getMessage()).thenReturn(yetAnotherWatchKey); when(releaseMessageService .findLatestReleaseMessagesGroupByMessages(Sets.newHashSet(watchKeysMap.values()))) .thenReturn(Lists.newArrayList(someReleaseMessage, yetAnotherReleaseMessage)); String notificationAsString = transformApolloConfigNotificationsToString(defaultNamespace, someNotificationId, somePublicNamespace, someNotificationId); DeferredResult<ResponseEntity<List<ApolloConfigNotification>>> deferredResult = controller .pollNotification(someAppId, someCluster, notificationAsString, someDataCenter, someClientIp); ResponseEntity<List<ApolloConfigNotification>> result = (ResponseEntity<List<ApolloConfigNotification>>) deferredResult.getResult(); assertEquals(HttpStatus.OK, result.getStatusCode()); assertEquals(1, result.getBody().size()); assertEquals(somePublicNamespace, result.getBody().get(0).getNamespaceName()); assertEquals(notificationId, result.getBody().get(0).getNotificationId()); ApolloNotificationMessages notificationMessages = result.getBody().get(0).getMessages(); assertEquals(2, notificationMessages.getDetails().size()); assertEquals(notificationId, notificationMessages.get(anotherWatchKey).longValue()); assertEquals(yetAnotherNotificationId, notificationMessages.get(yetAnotherWatchKey).longValue()); }