@Ignore("Publisher#flatMapMerge required https: @Test public void multipleSrvMultipleADiscover() throws InterruptedException { final String domain = "mysvc.apple.com"; final String targetDomain1 = "target1.mysvc.apple.com"; final String targetDomain2 = "target2.mysvc.apple.com"; final int targetPort1 = 9876; final int targetPort2 = 9878; recordStore.addSrvResponse(domain, targetDomain1, 10, 10, targetPort1); recordStore.addSrvResponse(domain, targetDomain2, 10, 10, targetPort2); recordStore.addResponse(targetDomain1, A, nextIp(), nextIp()); recordStore.addResponse(targetDomain2, A, nextIp(), nextIp()); final int expectedActiveCount = 4; final int expectedInactiveCount = 0; CountDownLatch latch = new CountDownLatch(expectedActiveCount + expectedInactiveCount); AtomicReference<Throwable> throwableRef = new AtomicReference<>(); Publisher<ServiceDiscovererEvent<InetSocketAddress>> publisher = client.dnsSrvQuery(domain) .flatMapConcatIterable(identity()); ServiceDiscovererTestSubscriber<InetSocketAddress> 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)); }
@Override public Publisher<Collection<ServiceDiscovererEvent<InetSocketAddress>>> dnsSrvQuery(final String serviceName) { requireNonNull(serviceName); return defer(() -> { final Map<HostAndPort, ARecordPublisher> aRecordMap = new HashMap<>(8); final DnsDiscoveryObserver discoveryObserver = newDiscoveryObserver(serviceName); return new SrvRecordPublisher(serviceName, discoveryObserver).flatMapConcatIterable(identity()) .flatMapMergeSingle(srvEvent -> { assertInEventloop(); if (srvEvent.isAvailable()) { final ARecordPublisher aPublisher = new ARecordPublisher(false, srvEvent.address().hostName(), discoveryObserver); final ARecordPublisher prevAPublisher = aRecordMap.putIfAbsent(srvEvent.address(), aPublisher); if (prevAPublisher != null) { return failed(new IllegalStateException("Only 1 A* record per SRV record is supported. " + srvEvent.address() + " corresponding to SRV name " + serviceName + " had a pre-existing A* record:" + prevAPublisher.name + " when new A* record arrived: " + aPublisher.name)); } return srvARecordPubToSingle(aPublisher, srvEvent, serviceName); } else { final ARecordPublisher aPublisher = aRecordMap.remove(srvEvent.address()); if (aPublisher != null) { final List<ServiceDiscovererEvent<InetAddress>> list = aPublisher.close0(); if (list != null) { return srvARecordPubToSingle(Publisher.from(list), srvEvent, serviceName); } } return failed(new IllegalStateException("Received an SRV inactive event for " + srvEvent.address() + " corresponding to SRV name " + serviceName + " but failed to find the corresponding A* record Publisher.")); } }); }); }
DefaultDnsClient implements DnsClient { @Override public Publisher<Collection<ServiceDiscovererEvent<InetSocketAddress>>> dnsSrvQuery(final String serviceName) { requireNonNull(serviceName); return defer(() -> { final Map<HostAndPort, ARecordPublisher> aRecordMap = new HashMap<>(8); final DnsDiscoveryObserver discoveryObserver = newDiscoveryObserver(serviceName); return new SrvRecordPublisher(serviceName, discoveryObserver).flatMapConcatIterable(identity()) .flatMapMergeSingle(srvEvent -> { assertInEventloop(); if (srvEvent.isAvailable()) { final ARecordPublisher aPublisher = new ARecordPublisher(false, srvEvent.address().hostName(), discoveryObserver); final ARecordPublisher prevAPublisher = aRecordMap.putIfAbsent(srvEvent.address(), aPublisher); if (prevAPublisher != null) { return failed(new IllegalStateException("Only 1 A* record per SRV record is supported. " + srvEvent.address() + " corresponding to SRV name " + serviceName + " had a pre-existing A* record:" + prevAPublisher.name + " when new A* record arrived: " + aPublisher.name)); } return srvARecordPubToSingle(aPublisher, srvEvent, serviceName); } else { final ARecordPublisher aPublisher = aRecordMap.remove(srvEvent.address()); if (aPublisher != null) { final List<ServiceDiscovererEvent<InetAddress>> list = aPublisher.close0(); if (list != null) { return srvARecordPubToSingle(Publisher.from(list), srvEvent, serviceName); } } return failed(new IllegalStateException("Received an SRV inactive event for " + srvEvent.address() + " corresponding to SRV name " + serviceName + " but failed to find the corresponding A* record Publisher.")); } }); }); } }
DefaultDnsClient implements DnsClient { @Override public Publisher<Collection<ServiceDiscovererEvent<InetSocketAddress>>> dnsSrvQuery(final String serviceName) { requireNonNull(serviceName); return defer(() -> { final Map<HostAndPort, ARecordPublisher> aRecordMap = new HashMap<>(8); final DnsDiscoveryObserver discoveryObserver = newDiscoveryObserver(serviceName); return new SrvRecordPublisher(serviceName, discoveryObserver).flatMapConcatIterable(identity()) .flatMapMergeSingle(srvEvent -> { assertInEventloop(); if (srvEvent.isAvailable()) { final ARecordPublisher aPublisher = new ARecordPublisher(false, srvEvent.address().hostName(), discoveryObserver); final ARecordPublisher prevAPublisher = aRecordMap.putIfAbsent(srvEvent.address(), aPublisher); if (prevAPublisher != null) { return failed(new IllegalStateException("Only 1 A* record per SRV record is supported. " + srvEvent.address() + " corresponding to SRV name " + serviceName + " had a pre-existing A* record:" + prevAPublisher.name + " when new A* record arrived: " + aPublisher.name)); } return srvARecordPubToSingle(aPublisher, srvEvent, serviceName); } else { final ARecordPublisher aPublisher = aRecordMap.remove(srvEvent.address()); if (aPublisher != null) { final List<ServiceDiscovererEvent<InetAddress>> list = aPublisher.close0(); if (list != null) { return srvARecordPubToSingle(Publisher.from(list), srvEvent, serviceName); } } return failed(new IllegalStateException("Received an SRV inactive event for " + srvEvent.address() + " corresponding to SRV name " + serviceName + " but failed to find the corresponding A* record Publisher.")); } }); }); } 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<InetSocketAddress>>> dnsSrvQuery(final String serviceName) { requireNonNull(serviceName); return defer(() -> { final Map<HostAndPort, ARecordPublisher> aRecordMap = new HashMap<>(8); final DnsDiscoveryObserver discoveryObserver = newDiscoveryObserver(serviceName); return new SrvRecordPublisher(serviceName, discoveryObserver).flatMapConcatIterable(identity()) .flatMapMergeSingle(srvEvent -> { assertInEventloop(); if (srvEvent.isAvailable()) { final ARecordPublisher aPublisher = new ARecordPublisher(false, srvEvent.address().hostName(), discoveryObserver); final ARecordPublisher prevAPublisher = aRecordMap.putIfAbsent(srvEvent.address(), aPublisher); if (prevAPublisher != null) { return failed(new IllegalStateException("Only 1 A* record per SRV record is supported. " + srvEvent.address() + " corresponding to SRV name " + serviceName + " had a pre-existing A* record:" + prevAPublisher.name + " when new A* record arrived: " + aPublisher.name)); } return srvARecordPubToSingle(aPublisher, srvEvent, serviceName); } else { final ARecordPublisher aPublisher = aRecordMap.remove(srvEvent.address()); if (aPublisher != null) { final List<ServiceDiscovererEvent<InetAddress>> list = aPublisher.close0(); if (list != null) { return srvARecordPubToSingle(Publisher.from(list), srvEvent, serviceName); } } return failed(new IllegalStateException("Received an SRV inactive event for " + srvEvent.address() + " corresponding to SRV name " + serviceName + " but failed to find the corresponding A* record Publisher.")); } }); }); } 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<InetSocketAddress>>> dnsSrvQuery(final String serviceName) { requireNonNull(serviceName); return defer(() -> { final Map<HostAndPort, ARecordPublisher> aRecordMap = new HashMap<>(8); final DnsDiscoveryObserver discoveryObserver = newDiscoveryObserver(serviceName); return new SrvRecordPublisher(serviceName, discoveryObserver).flatMapConcatIterable(identity()) .flatMapMergeSingle(srvEvent -> { assertInEventloop(); if (srvEvent.isAvailable()) { final ARecordPublisher aPublisher = new ARecordPublisher(false, srvEvent.address().hostName(), discoveryObserver); final ARecordPublisher prevAPublisher = aRecordMap.putIfAbsent(srvEvent.address(), aPublisher); if (prevAPublisher != null) { return failed(new IllegalStateException("Only 1 A* record per SRV record is supported. " + srvEvent.address() + " corresponding to SRV name " + serviceName + " had a pre-existing A* record:" + prevAPublisher.name + " when new A* record arrived: " + aPublisher.name)); } return srvARecordPubToSingle(aPublisher, srvEvent, serviceName); } else { final ARecordPublisher aPublisher = aRecordMap.remove(srvEvent.address()); if (aPublisher != null) { final List<ServiceDiscovererEvent<InetAddress>> list = aPublisher.close0(); if (list != null) { return srvARecordPubToSingle(Publisher.from(list), srvEvent, serviceName); } } return failed(new IllegalStateException("Received an SRV inactive event for " + srvEvent.address() + " corresponding to SRV name " + serviceName + " but failed to find the corresponding A* record Publisher.")); } }); }); } 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();  }