NotificationController implements ReleaseMessageListener { @GetMapping public DeferredResult<ResponseEntity<ApolloConfigNotification>> pollNotification( @RequestParam(value = "appId") String appId, @RequestParam(value = "cluster") String cluster, @RequestParam(value = "namespace", defaultValue = ConfigConsts.NAMESPACE_APPLICATION) String namespace, @RequestParam(value = "dataCenter", required = false) String dataCenter, @RequestParam(value = "notificationId", defaultValue = "-1") long notificationId, @RequestParam(value = "ip", required = false) String clientIp) { namespace = namespaceUtil.filterNamespaceName(namespace); Set<String> watchedKeys = watchKeysUtil.assembleAllWatchKeys(appId, cluster, namespace, dataCenter); DeferredResult<ResponseEntity<ApolloConfigNotification>> deferredResult = new DeferredResult<>(TIMEOUT, NOT_MODIFIED_RESPONSE); ReleaseMessage latest = releaseMessageService.findLatestReleaseMessageForMessages(watchedKeys); entityManagerUtil.closeEntityManager(); if (latest != null && latest.getId() != notificationId) { deferredResult.setResult(new ResponseEntity<>( new ApolloConfigNotification(namespace, latest.getId()), HttpStatus.OK)); } else { for (String key : watchedKeys) { this.deferredResults.put(key, deferredResult); } deferredResult .onTimeout(() -> logWatchedKeys(watchedKeys, "Apollo.LongPoll.TimeOutKeys")); deferredResult.onCompletion(() -> { for (String key : watchedKeys) { deferredResults.remove(key, deferredResult); } logWatchedKeys(watchedKeys, "Apollo.LongPoll.CompletedKeys"); }); logWatchedKeys(watchedKeys, "Apollo.LongPoll.RegisteredKeys"); logger.debug("Listening {} from appId: {}, cluster: {}, namespace: {}, datacenter: {}", watchedKeys, appId, cluster, namespace, dataCenter); } return deferredResult; } NotificationController(
      final WatchKeysUtil watchKeysUtil,
      final ReleaseMessageServiceWithCache releaseMessageService,
      final EntityManagerUtil entityManagerUtil,
      final NamespaceUtil namespaceUtil); @GetMapping DeferredResult<ResponseEntity<ApolloConfigNotification>> pollNotification(
      @RequestParam(value = "appId") String appId,
      @RequestParam(value = "cluster") String cluster,
      @RequestParam(value = "namespace", defaultValue = ConfigConsts.NAMESPACE_APPLICATION) String namespace,
      @RequestParam(value = "dataCenter", required = false) String dataCenter,
      @RequestParam(value = "notificationId", defaultValue = "-1") long notificationId,
      @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"; Set<String> watchKeys = Sets.newHashSet(someWatchKey, anotherWatchKey); when(watchKeysUtil .assembleAllWatchKeys(someAppId, someCluster, defaultNamespace, someDataCenter)).thenReturn( watchKeys); DeferredResult<ResponseEntity<ApolloConfigNotification>> deferredResult = controller .pollNotification(someAppId, someCluster, defaultNamespace, someDataCenter, someNotificationId, someClientIp); assertEquals(watchKeys.size(), deferredResults.size()); for (String watchKey : watchKeys) { assertTrue(deferredResults.get(watchKey).contains(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"; Set<String> watchKeys = Sets.newHashSet(someWatchKey, anotherWatchKey); when(watchKeysUtil .assembleAllWatchKeys(someAppId, someCluster, defaultNamespace, someDataCenter)).thenReturn( watchKeys); DeferredResult<ResponseEntity<ApolloConfigNotification>> deferredResult = controller .pollNotification(someAppId, someCluster, namespace, someDataCenter, someNotificationId, someClientIp); assertEquals(watchKeys.size(), deferredResults.size()); for (String watchKey : watchKeys) { assertTrue(deferredResults.get(watchKey).contains(deferredResult)); } }
@Test public void testPollNotificationWithSomeNamespaceAsFile() throws Exception { String namespace = String.format("someNamespace.xml"); when(namespaceUtil.filterNamespaceName(namespace)).thenReturn(namespace); String someWatchKey = "someKey"; Set<String> watchKeys = Sets.newHashSet(someWatchKey); when(watchKeysUtil .assembleAllWatchKeys(someAppId, someCluster, namespace, someDataCenter)) .thenReturn( watchKeys); DeferredResult<ResponseEntity<ApolloConfigNotification>> deferredResult = controller .pollNotification(someAppId, someCluster, namespace, someDataCenter, someNotificationId, someClientIp); assertEquals(watchKeys.size(), deferredResults.size()); for (String watchKey : watchKeys) { assertTrue(deferredResults.get(watchKey).contains(deferredResult)); } }
@Test public void testPollNotificationWithDefaultNamespaceWithNotificationIdOutDated() throws Exception { long notificationId = someNotificationId + 1; ReleaseMessage someReleaseMessage = mock(ReleaseMessage.class); String someWatchKey = "someKey"; Set<String> watchKeys = Sets.newHashSet(someWatchKey); when(watchKeysUtil .assembleAllWatchKeys(someAppId, someCluster, defaultNamespace, someDataCenter)) .thenReturn( watchKeys); when(someReleaseMessage.getId()).thenReturn(notificationId); when(releaseMessageService.findLatestReleaseMessageForMessages(watchKeys)) .thenReturn(someReleaseMessage); DeferredResult<ResponseEntity<ApolloConfigNotification>> deferredResult = controller .pollNotification(someAppId, someCluster, defaultNamespace, someDataCenter, someNotificationId, someClientIp); ResponseEntity<ApolloConfigNotification> result = (ResponseEntity<ApolloConfigNotification>) deferredResult.getResult(); assertEquals(HttpStatus.OK, result.getStatusCode()); assertEquals(defaultNamespace, result.getBody().getNamespaceName()); assertEquals(notificationId, result.getBody().getNotificationId()); }