@Test public void testObtainServerListByKvStore() throws Exception { List<Node> list1 = Arrays.asList( new Node("node1", "node1/8080/0", 0), new Node("node2", "node2/8080/0", 0), new Node("node3", "node3/8080/0", 0) ); List<Node> list2 = Arrays.asList( new Node("node2", "node2/8080/0", 0), new Node("node3", "node3/8080/0", 0), new Node("node4", "node4/8080/0", 0) ); MockKvStore store = new MockKvStore() { int c = 0; @Override public List<Node> list(String prefix) throws Exception { if (c++ % 2 == 0) return list1; else return list2; } }; RegisteredServerDynamicList dynamicList = new RegisteredServerDynamicList(store, "abc"); List<RegisteredServer> serverList1 = dynamicList.obtainServerListByKvStore(); assertEquals(3, serverList1.size()); for (int i = 1; i < 4; i++) { assertEquals("node" + i, serverList1.get(i - 1).getHost()); } List<RegisteredServer> serverList2 = dynamicList.obtainServerListByKvStore(); assertEquals(3, serverList2.size()); for (int i = 2; i < 5; i++) { assertEquals("node" + i, serverList2.get(i - 2).getHost()); } Map<String, ProviderInfo> map = dynamicList.getProviderMap(); assertEquals(3, map.size()); assertTrue(map.keySet().contains("node4")); assertFalse(map.keySet().contains("node1")); }
List<RegisteredServer> obtainServerListByKvStore() { Set<String> newKeys = new HashSet<>(); Consumer<Node> collectNewNodesToSet = node -> newKeys.add(node.getKey()); try { List<Node> nodeInfoList = store.list(parentNode); Stream<ProviderInfo> stream = nodeInfoList .stream() .peek(collectNewNodesToSet) .map(node -> providerMap.computeIfAbsent( node.getKey(), k -> MapperUtil.node2Provider(node))); providerMap.keySet() .parallelStream() .filter(k -> !newKeys.contains(k)) .forEach(k -> providerMap.remove(k)); return map2ServerList(stream); } catch (Exception e) { log.warn(String.format("failed to obtain list of servers: %s", parentNode), e); return Lists.newArrayList(); } }
RegisteredServerDynamicList implements ServerList<RegisteredServer> { List<RegisteredServer> obtainServerListByKvStore() { Set<String> newKeys = new HashSet<>(); Consumer<Node> collectNewNodesToSet = node -> newKeys.add(node.getKey()); try { List<Node> nodeInfoList = store.list(parentNode); Stream<ProviderInfo> stream = nodeInfoList .stream() .peek(collectNewNodesToSet) .map(node -> providerMap.computeIfAbsent( node.getKey(), k -> MapperUtil.node2Provider(node))); providerMap.keySet() .parallelStream() .filter(k -> !newKeys.contains(k)) .forEach(k -> providerMap.remove(k)); return map2ServerList(stream); } catch (Exception e) { log.warn(String.format("failed to obtain list of servers: %s", parentNode), e); return Lists.newArrayList(); } } }
RegisteredServerDynamicList implements ServerList<RegisteredServer> { List<RegisteredServer> obtainServerListByKvStore() { Set<String> newKeys = new HashSet<>(); Consumer<Node> collectNewNodesToSet = node -> newKeys.add(node.getKey()); try { List<Node> nodeInfoList = store.list(parentNode); Stream<ProviderInfo> stream = nodeInfoList .stream() .peek(collectNewNodesToSet) .map(node -> providerMap.computeIfAbsent( node.getKey(), k -> MapperUtil.node2Provider(node))); providerMap.keySet() .parallelStream() .filter(k -> !newKeys.contains(k)) .forEach(k -> providerMap.remove(k)); return map2ServerList(stream); } catch (Exception e) { log.warn(String.format("failed to obtain list of servers: %s", parentNode), e); return Lists.newArrayList(); } } RegisteredServerDynamicList(KVStore store, Application application);  RegisteredServerDynamicList(KVStore store, String serverGroup); }
RegisteredServerDynamicList implements ServerList<RegisteredServer> { List<RegisteredServer> obtainServerListByKvStore() { Set<String> newKeys = new HashSet<>(); Consumer<Node> collectNewNodesToSet = node -> newKeys.add(node.getKey()); try { List<Node> nodeInfoList = store.list(parentNode); Stream<ProviderInfo> stream = nodeInfoList .stream() .peek(collectNewNodesToSet) .map(node -> providerMap.computeIfAbsent( node.getKey(), k -> MapperUtil.node2Provider(node))); providerMap.keySet() .parallelStream() .filter(k -> !newKeys.contains(k)) .forEach(k -> providerMap.remove(k)); return map2ServerList(stream); } catch (Exception e) { log.warn(String.format("failed to obtain list of servers: %s", parentNode), e); return Lists.newArrayList(); } } RegisteredServerDynamicList(KVStore store, Application application);  RegisteredServerDynamicList(KVStore store, String serverGroup); ServerListUpdater getKvStoreBasedUpdater(); ServerListUpdater getKvStoreBasedUpdater(ExecutorService executorService); Collection<ProviderInfo> getCachedListOfServers(); @Override List<RegisteredServer> getInitialListOfServers(); @Override List<RegisteredServer> getUpdatedListOfServers(); }
RegisteredServerDynamicList implements ServerList<RegisteredServer> { List<RegisteredServer> obtainServerListByKvStore() { Set<String> newKeys = new HashSet<>(); Consumer<Node> collectNewNodesToSet = node -> newKeys.add(node.getKey()); try { List<Node> nodeInfoList = store.list(parentNode); Stream<ProviderInfo> stream = nodeInfoList .stream() .peek(collectNewNodesToSet) .map(node -> providerMap.computeIfAbsent( node.getKey(), k -> MapperUtil.node2Provider(node))); providerMap.keySet() .parallelStream() .filter(k -> !newKeys.contains(k)) .forEach(k -> providerMap.remove(k)); return map2ServerList(stream); } catch (Exception e) { log.warn(String.format("failed to obtain list of servers: %s", parentNode), e); return Lists.newArrayList(); } } RegisteredServerDynamicList(KVStore store, Application application);  RegisteredServerDynamicList(KVStore store, String serverGroup); ServerListUpdater getKvStoreBasedUpdater(); ServerListUpdater getKvStoreBasedUpdater(ExecutorService executorService); Collection<ProviderInfo> getCachedListOfServers(); @Override List<RegisteredServer> getInitialListOfServers(); @Override List<RegisteredServer> getUpdatedListOfServers();  }