@Test public void repeatDiscoverMultipleHosts() throws Exception { recordStore.addResponse("apple.com", A, nextIp()) .defaultResponse("apple.com", A, nextIp()) .addResponse("servicetalk.io", A, nextIp()) .defaultResponse("servicetalk.io", A, nextIp()); final int expectedAppleActiveCount = 2; final int expectedAppleInactiveCount = 1; final int expectedStActiveCount = 2; final int expectedStInactiveCount = 1; CountDownLatch appleLatch = new CountDownLatch(expectedAppleActiveCount + expectedAppleInactiveCount); CountDownLatch stLatch = new CountDownLatch(expectedStActiveCount + expectedStInactiveCount); AtomicReference<Throwable> throwableRef = new AtomicReference<>(); Publisher<ServiceDiscovererEvent<InetAddress>> applePublisher = client.dnsQuery("apple.com") .flatMapConcatIterable(identity()); Publisher<ServiceDiscovererEvent<InetAddress>> stPublisher = client.dnsQuery("servicetalk.io") .flatMapConcatIterable(identity()); ServiceDiscovererTestSubscriber<InetAddress> appleSubscriber = new ServiceDiscovererTestSubscriber<>(appleLatch, throwableRef, Long.MAX_VALUE); ServiceDiscovererTestSubscriber<InetAddress> stSubscriber = new ServiceDiscovererTestSubscriber<>(stLatch, throwableRef, Long.MAX_VALUE); toSource(applePublisher).subscribe(appleSubscriber); toSource(stPublisher).subscribe(stSubscriber); appleLatch.await(); stLatch.await(); assertNull(throwableRef.get()); assertThat(appleSubscriber.activeCount(), equalTo(expectedAppleActiveCount)); assertThat(appleSubscriber.inactiveCount(), equalTo(expectedAppleInactiveCount)); assertThat(stSubscriber.activeCount(), equalTo(expectedStActiveCount)); assertThat(stSubscriber.inactiveCount(), equalTo(expectedStInactiveCount)); }
@Override public Publisher<Collection<ServiceDiscovererEvent<InetAddress>>> dnsQuery(final String address) { requireNonNull(address); return defer(() -> new ARecordPublisher(true, address, newDiscoveryObserver(address))); }
DefaultDnsClient implements DnsClient { @Override public Publisher<Collection<ServiceDiscovererEvent<InetAddress>>> dnsQuery(final String address) { requireNonNull(address); return defer(() -> new ARecordPublisher(true, address, newDiscoveryObserver(address))); } }
DefaultDnsClient implements DnsClient { @Override public Publisher<Collection<ServiceDiscovererEvent<InetAddress>>> dnsQuery(final String address) { requireNonNull(address); return defer(() -> new ARecordPublisher(true, address, newDiscoveryObserver(address))); } DefaultDnsClient(final IoExecutor ioExecutor, final int minTTL,
                     @Nullable final Integer ndots,
                     @Nullable final Boolean optResourceEnabled, @Nullable final Duration queryTimeout,
                     @Nullable final DnsResolverAddressTypes dnsResolverAddressTypes,
                     @Nullable final DnsServerAddressStreamProvider dnsServerAddressStreamProvider,
                     @Nullable final DnsServiceDiscovererObserver observer); }
DefaultDnsClient implements DnsClient { @Override public Publisher<Collection<ServiceDiscovererEvent<InetAddress>>> dnsQuery(final String address) { requireNonNull(address); return defer(() -> new ARecordPublisher(true, address, newDiscoveryObserver(address))); } DefaultDnsClient(final IoExecutor ioExecutor, final int minTTL,
                     @Nullable final Integer ndots,
                     @Nullable final Boolean optResourceEnabled, @Nullable final Duration queryTimeout,
                     @Nullable final DnsResolverAddressTypes dnsResolverAddressTypes,
                     @Nullable final DnsServerAddressStreamProvider dnsServerAddressStreamProvider,
                     @Nullable final DnsServiceDiscovererObserver observer); @Override Publisher<Collection<ServiceDiscovererEvent<InetAddress>>> dnsQuery(final String address); @Override Publisher<Collection<ServiceDiscovererEvent<InetSocketAddress>>> dnsSrvQuery(final String serviceName); @Override Completable onClose(); @Override Completable closeAsync(); @Override Completable closeAsyncGracefully(); }
DefaultDnsClient implements DnsClient { @Override public Publisher<Collection<ServiceDiscovererEvent<InetAddress>>> dnsQuery(final String address) { requireNonNull(address); return defer(() -> new ARecordPublisher(true, address, newDiscoveryObserver(address))); } DefaultDnsClient(final IoExecutor ioExecutor, final int minTTL,
                     @Nullable final Integer ndots,
                     @Nullable final Boolean optResourceEnabled, @Nullable final Duration queryTimeout,
                     @Nullable final DnsResolverAddressTypes dnsResolverAddressTypes,
                     @Nullable final DnsServerAddressStreamProvider dnsServerAddressStreamProvider,
                     @Nullable final DnsServiceDiscovererObserver observer); @Override Publisher<Collection<ServiceDiscovererEvent<InetAddress>>> dnsQuery(final String address); @Override Publisher<Collection<ServiceDiscovererEvent<InetSocketAddress>>> dnsSrvQuery(final String serviceName); @Override Completable onClose(); @Override Completable closeAsync(); @Override Completable closeAsyncGracefully();  }