retryWithTcp() — netty Function Reference

Source Code

resolver-dns/src/main/java/io/netty/resolver/dns/DnsQueryContext.java lines 349–428

    private boolean retryWithTcp(final Object originalResult) {
        if (socketBootstrap == null) {
            return false;
        }

        socketBootstrap.connect(nameServerAddr).addListener((ChannelFutureListener) future -> {
            if (!future.isSuccess()) {
                logger.debug("{} Unable to fallback to TCP [{}: {}]",
                        future.channel(), id, nameServerAddr, future.cause());
                // TCP fallback failed, just use the truncated response or error.
                finishOriginal(originalResult, future);
                return;
            }
            final Channel tcpCh = future.channel();
            Promise<AddressedEnvelope<DnsResponse, InetSocketAddress>> promise =
                    tcpCh.eventLoop().newPromise();
            final TcpDnsQueryContext tcpCtx = new TcpDnsQueryContext(tcpCh,
                    (InetSocketAddress) tcpCh.remoteAddress(), queryContextManager, queryLifecycleObserver, 0,
                    recursionDesired, queryTimeoutMillis, question(), additionals, promise);
            tcpCh.pipeline().addLast(TCP_ENCODER);
            tcpCh.pipeline().addLast(new TcpDnsResponseDecoder());
            tcpCh.pipeline().addLast(new ChannelInboundHandlerAdapter() {
                @Override
                public void channelRead(ChannelHandlerContext ctx, Object msg) {
                    Channel tcpCh = ctx.channel();
                    DnsResponse response = (DnsResponse) msg;
                    int queryId = response.id();

                    if (logger.isDebugEnabled()) {
                        logger.debug("{} RECEIVED: TCP [{}: {}], {}", tcpCh, queryId,
                                tcpCh.remoteAddress(), response);
                    }

                    DnsQueryContext foundCtx = queryContextManager.get(nameServerAddr, queryId);
                    if (foundCtx != null && foundCtx.isDone()) {
                        logger.debug("{} Received a DNS response for a query that was timed out or cancelled " +
                                ": TCP [{}: {}]", tcpCh, queryId, nameServerAddr);
                        response.release();
                    } else if (foundCtx == tcpCtx) {
                        tcpCtx.finishSuccess(new AddressedEnvelopeAdapter(
                                (InetSocketAddress) ctx.channel().remoteAddress(),
                                (InetSocketAddress) ctx.channel().localAddress(),
                                response), false);
                    } else {
                        response.release();
                        tcpCtx.finishFailure("Received TCP DNS response with unexpected ID", null, false);
                        if (logger.isDebugEnabled()) {
                            logger.debug("{} Received a DNS response with an unexpected ID: TCP [{}: {}]",
                                    tcpCh, queryId, tcpCh.remoteAddress());
                        }
                    }
                }

                @Override
                public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
                    if (tcpCtx.finishFailure(
                            "TCP fallback error", cause, false) && logger.isDebugEnabled()) {
                        logger.debug("{} Error during processing response: TCP [{}: {}]",
                                ctx.channel(), id,
                                ctx.channel().remoteAddress(), cause);
                    }
                }
            });

            promise.addListener(
                    (FutureListener<AddressedEnvelope<DnsResponse, InetSocketAddress>>) future1 -> {
                        if (future1.isSuccess()) {
                            finishSuccess(future1.getNow(), false);
                            // Release the original result.
                            ReferenceCountUtil.release(originalResult);
                        } else {
                            // TCP fallback failed, just use the truncated response or error.
                            finishOriginal(originalResult, future1);
                        }
                        tcpCh.close();
                    });
            tcpCtx.writeQuery(true);
        });
        return true;
    }