doQuery() — netty Function Reference

Source Code

resolver-dns/src/main/java/io/netty/resolver/dns/DnsNameResolver.java lines 1389–1471

    @SuppressWarnings("unchecked")
    final Future<AddressedEnvelope<DnsResponse, InetSocketAddress>> doQuery(
            Channel channel,
            InetSocketAddress nameServerAddr, DnsQuestion question,
            final DnsQueryLifecycleObserver queryLifecycleObserver,
            DnsRecord[] additionals, boolean flush,
            Promise<AddressedEnvelope<? extends DnsResponse, InetSocketAddress>> promise) {
        final Promise<AddressedEnvelope<DnsResponse, InetSocketAddress>> castPromise = cast(
                checkNotNull(promise, "promise"));
        final int payloadSize = isOptResourceEnabled() ? maxPayloadSize() : 0;

        if (inflightLookups != null && (additionals == null || additionals.length == 0)) {
            Promise<AddressedEnvelope<? extends DnsResponse, InetSocketAddress>> inflight =
                    inflightLookups.get(question);
            if (inflight != null) {
                // We have a query / response inflight, let's just cascade on it to reduce the network traffic.
                inflight.addListener(f -> {
                    if (f.isSuccess()) {
                        // Notify the observer and after that the promise
                        queryLifecycleObserver.querySucceed();
                        AddressedEnvelope<? extends DnsResponse, InetSocketAddress> result =
                                (AddressedEnvelope<? extends DnsResponse, InetSocketAddress>) f.getNow();

                        // Retain the result as the listener on the promise is responsible to release it.
                        ReferenceCountUtil.retain(result);
                        promise.setSuccess(result);
                    } else {
                        Throwable cause = f.cause();
                        if (isTimeoutError(cause)) {
                            doQueryNow(channel, nameServerAddr, question, queryLifecycleObserver,
                                    additionals, flush, payloadSize, castPromise);
                        } else {
                            // Notify the observer and after that the promise
                            queryLifecycleObserver.queryFailed(cause);
                            promise.setFailure(cause);
                        }
                    }
                });
                return castPromise;
            } else if (inflightLookups.size() < maxNumConsolidation) {
                // Create a new promise as we need to ensure we are the first that will add a listener to it to retain
                // the result before anyone can release it.
                Promise<AddressedEnvelope<? extends DnsResponse, InetSocketAddress>> newPromise =
                        executor().newPromise();
                Promise<AddressedEnvelope<? extends DnsResponse, InetSocketAddress>> old =
                        inflightLookups.put(question, newPromise);
                assert old == null;

                newPromise.addListener(f -> {
                    // Remove the promise and add another listener to it that will call release() on the result.
                    // As the execution of the listeners is guaranteed to be in the same order as how these were added
                    // we know that all previous added listeners had a chance to handle the result already.
                    Promise<AddressedEnvelope<? extends DnsResponse, InetSocketAddress>> p =
                            inflightLookups.remove(question);
                    assert p == newPromise;

                    if (f.isSuccess()) {
                        // On success we need to retain the result so listeners that are added after this one
                        // will still be able to use the result.
                        AddressedEnvelope<? extends DnsResponse, InetSocketAddress> result =
                                (AddressedEnvelope<? extends DnsResponse, InetSocketAddress>) f.getNow();
                        ReferenceCountUtil.retain(result);
                        promise.setSuccess(result);
                    } else {
                        promise.setFailure(f.cause());
                    }

                    p.addListener(RELEASE_LISTENER);
                });

                doQueryNow(channel, nameServerAddr, question, queryLifecycleObserver,
                        additionals, flush, payloadSize, cast(newPromise));

                // Return the original castPromise which will be notified by the newPromise that we used above.
                // This was it's impossible for the user to add any extra listeners to the newPromise itself, which
                // is needed to guarantee the correct life-cycle of the reference counted response.
                return castPromise;
            }
        }
        doQueryNow(channel, nameServerAddr, question, queryLifecycleObserver,
                additionals, flush, payloadSize, castPromise);