@Test public void repeatDiscoverMultiTtl() throws InterruptedException { final String ipA1 = nextIp(); final String ipA2 = nextIp(); final String ipB1 = nextIp(); final String ipB2 = nextIp(); AtomicLong firstTime = new AtomicLong(); AtomicLong secondTime = new AtomicLong(); recordStore .addResponse("apple.com", A, () -> { firstTime.set(System.currentTimeMillis()); return asList(createRecord("apple.com", A, 1, ipA1), createRecord("apple.com", A, 10, ipA2)); }) .defaultResponse("apple.com", A, () -> { secondTime.set(System.currentTimeMillis()); return asList(createRecord("apple.com", A, 10, ipB1), createRecord("apple.com", A, 10, ipB2)); }); final int expectedActiveCount = 4; final int expectedInactiveCount = 2; CountDownLatch latch = new CountDownLatch(expectedActiveCount + expectedInactiveCount); Publisher<ServiceDiscovererEvent<InetAddress>> publisher = client.dnsQuery("apple.com") .flatMapConcatIterable(identity()); final TestSubscriber subscriber = new TestSubscriber(latch); toSource(publisher).subscribe(subscriber); latch.await(); assertNull(subscriber.throwableRef.get()); assertThat(new HashSet<>(subscriber.activeEventAddresses), equalTo(new HashSet<>(asList(ipA1, ipA2, ipB1, ipB2)))); assertThat(subscriber.activeEventAddresses.size(), equalTo(expectedActiveCount)); assertThat(new HashSet<>(subscriber.inactiveEventAddresses), equalTo(new HashSet<>(asList(ipA1, ipA2)))); assertThat(subscriber.inactiveEventAddresses.size(), equalTo(expectedInactiveCount)); long timeBetweenQueries = secondTime.get() - firstTime.get(); assertThat(timeBetweenQueries, greaterThanOrEqualTo(1000L)); assertThat(timeBetweenQueries, lessThan(10_000L)); }
@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();  }