@SuppressWarnings("unchecked") @Test public void handleDiscoveryEvents() { assertThat(lb.activeAddresses(), is(empty())); sendServiceDiscoveryEvents(upEvent("address-1")); assertThat(lb.activeAddresses(), hasItems( both(hasProperty("key", is("address-1"))).and(hasProperty("value", is(empty()))))); sendServiceDiscoveryEvents(downEvent("address-1")); assertThat(lb.activeAddresses(), is(empty())); sendServiceDiscoveryEvents(upEvent("address-2")); assertThat(lb.activeAddresses(), hasItems( both(hasProperty("key", is("address-2"))).and(hasProperty("value", is(empty()))))); sendServiceDiscoveryEvents(downEvent("address-3")); assertThat(lb.activeAddresses(), hasItems( both(hasProperty("key", is("address-2"))).and(hasProperty("value", is(empty()))))); sendServiceDiscoveryEvents(upEvent("address-1")); assertThat(lb.activeAddresses(), hasItems( both(hasProperty("key", is("address-1"))).and(hasProperty("value", is(empty()))), both(hasProperty("key", is("address-2"))).and(hasProperty("value", is(empty()))))); sendServiceDiscoveryEvents(downEvent("address-1")); assertThat(lb.activeAddresses(), hasItems( both(hasProperty("key", is("address-2"))).and(hasProperty("value", is(empty()))))); sendServiceDiscoveryEvents(downEvent("address-2")); assertThat(lb.activeAddresses(), is(empty())); sendServiceDiscoveryEvents(downEvent("address-3")); assertThat(lb.activeAddresses(), is(empty())); sendServiceDiscoveryEvents(upEvent("address-1")); assertThat(lb.activeAddresses(), hasItems( both(hasProperty("key", is("address-1"))).and(hasProperty("value", is(empty()))))); serviceDiscoveryPublisher.onError(DELIBERATE_EXCEPTION); assertThat(lb.activeAddresses(), hasItems( both(hasProperty("key", is("address-1"))).and(hasProperty("value", is(empty()))))); }
List<Entry<ResolvedAddress, List<C>>> activeAddresses() { return activeHosts.stream().map(Host::asEntry).collect(toList()); }
RoundRobinLoadBalancer implements LoadBalancer<C> { List<Entry<ResolvedAddress, List<C>>> activeAddresses() { return activeHosts.stream().map(Host::asEntry).collect(toList()); } }
RoundRobinLoadBalancer implements LoadBalancer<C> { List<Entry<ResolvedAddress, List<C>>> activeAddresses() { return activeHosts.stream().map(Host::asEntry).collect(toList()); } RoundRobinLoadBalancer(final Publisher<? extends ServiceDiscovererEvent<ResolvedAddress>> eventPublisher,
                                  final ConnectionFactory<ResolvedAddress, ? extends C> connectionFactory); }
RoundRobinLoadBalancer implements LoadBalancer<C> { List<Entry<ResolvedAddress, List<C>>> activeAddresses() { return activeHosts.stream().map(Host::asEntry).collect(toList()); } RoundRobinLoadBalancer(final Publisher<? extends ServiceDiscovererEvent<ResolvedAddress>> eventPublisher,
                                  final ConnectionFactory<ResolvedAddress, ? extends C> connectionFactory); static RoundRobinLoadBalancerFactory<ResolvedAddress, C> newRoundRobinFactory(); @Override Single<C> selectConnection(Predicate<C> selector); @Override Publisher<Object> eventStream(); @Override Completable onClose(); @Override Completable closeAsync(); @Override Completable closeAsyncGracefully(); }
RoundRobinLoadBalancer implements LoadBalancer<C> { List<Entry<ResolvedAddress, List<C>>> activeAddresses() { return activeHosts.stream().map(Host::asEntry).collect(toList()); } RoundRobinLoadBalancer(final Publisher<? extends ServiceDiscovererEvent<ResolvedAddress>> eventPublisher,
                                  final ConnectionFactory<ResolvedAddress, ? extends C> connectionFactory); static RoundRobinLoadBalancerFactory<ResolvedAddress, C> newRoundRobinFactory(); @Override Single<C> selectConnection(Predicate<C> selector); @Override Publisher<Object> eventStream(); @Override Completable onClose(); @Override Completable closeAsync(); @Override Completable closeAsyncGracefully();  }