CuratorServiceGroupMonitor implements ServiceGroupMonitor { @Override public ImmutableSet<ServiceInstance> get() { return groupCache.getCurrentData().stream() .filter(cd -> memberSelector.test(ZKPaths.getNodeFromPath(cd.getPath()))) .map(this::extractServiceInstance) .filter(Optional::isPresent) .map(Optional::get) .collect(GuavaUtils.toImmutableSet()); } CuratorServiceGroupMonitor(
      PathChildrenCache groupCache,
      Predicate<String> memberSelector,
      Codec<ServiceInstance> codec); @Override void start(); @Override void close(); @Override ImmutableSet<ServiceInstance> get();  }
@Test public void testNoHosts() throws Exception { assertEquals(ImmutableSet.of(), getGroupMonitor().get()); startGroupMonitor(); assertEquals(ImmutableSet.of(), getGroupMonitor().get()); }
@Test public void testHostUpdates() throws Exception { startGroupMonitor(); ServiceInstance one = serviceInstance("one"); String onePath = createMember(one); ServiceInstance two = serviceInstance("two"); String twoPath = createMember(two); assertEquals(ImmutableSet.of(one, two), getGroupMonitor().get()); deleteChild(twoPath); assertEquals(ImmutableSet.of(one), getGroupMonitor().get()); deleteChild(onePath); ServiceInstance three = serviceInstance("three"); String threePath = createMember(three); assertEquals(ImmutableSet.of(three), getGroupMonitor().get()); deleteChild(threePath); assertEquals(ImmutableSet.of(), getGroupMonitor().get()); }
@Test public void testMixedNodes() throws Exception { startGroupMonitor(); String nonMemberPath = createNonMember(); assertEquals(ImmutableSet.of(), getGroupMonitor().get()); ServiceInstance member = serviceInstance("member"); String memberPath = createMember(member); assertEquals(ImmutableSet.of(member), getGroupMonitor().get()); deleteChild(memberPath); assertEquals(ImmutableSet.of(), getGroupMonitor().get()); deleteChild(nonMemberPath); assertEquals(ImmutableSet.of(), getGroupMonitor().get()); }
@Test public void testInvalidMemberNode() throws Exception { startGroupMonitor(); createMember(ThriftBinaryCodec.encode(serviceInstance("invalid"))); ServiceInstance member = serviceInstance("member"); createMember(member); assertEquals(ImmutableSet.of(member), getGroupMonitor().get()); }
@Test public void testStartBlocksOnInitialMembership() throws Exception { ServiceInstance one = serviceInstance("one"); createMember(one, false ); ServiceInstance two = serviceInstance("two"); createMember(two, false ); assertEquals(ImmutableSet.of(), getGroupMonitor().get()); startGroupMonitor(); assertEquals(ImmutableSet.of(one, two), getGroupMonitor().get()); }