@Test public void repeatDiscoverTtl() throws InterruptedException { AtomicLong firstTime = new AtomicLong(); AtomicLong secondTime = new AtomicLong(); recordStore .addResponse("apple.com", A, () -> { firstTime.set(System.currentTimeMillis()); return singletonList(createRecord("apple.com", A, 2, nextIp())); }) .defaultResponse("apple.com", A, () -> { secondTime.set(System.currentTimeMillis()); return singletonList(createRecord("apple.com", A, 2, nextIp())); }); final int expectedActiveCount = 2; final int expectedInactiveCount = 1; CountDownLatch latch = new CountDownLatch(expectedActiveCount + expectedInactiveCount); AtomicReference<Throwable> throwableRef = new AtomicReference<>(); Publisher<ServiceDiscovererEvent<InetAddress>> publisher = client.dnsQuery("apple.com") .flatMapConcatIterable(identity()); ServiceDiscovererTestSubscriber<InetAddress> subscriber = new ServiceDiscovererTestSubscriber<>(latch, throwableRef, Long.MAX_VALUE); toSource(publisher).subscribe(subscriber); latch.await(); assertNull(throwableRef.get()); assertThat(subscriber.activeCount(), equalTo(expectedActiveCount)); assertThat(subscriber.inactiveCount(), equalTo(expectedInactiveCount)); long timeBetweenQueries = secondTime.get() - firstTime.get(); assertThat(timeBetweenQueries, greaterThanOrEqualTo(2000L)); }
@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();  }