io.netty.handler.codec.dns.DefaultDnsQuestion #io.netty.handler.codec.dns.DnsRecordType源码实例Demo

@RequestMapping("/resolve")
public Mono<DnsResponse> query(@RequestParam("name") String name,
                               @RequestParam(name = "type", required = false, defaultValue = "A") String type,
                               @RequestParam(name = "cd", required = false, defaultValue = "false") boolean cd,
                               @RequestParam(name = "ct", required = false, defaultValue = "") String ct,
                               @RequestParam(name = "do", required = false, defaultValue = "false") boolean do1,
                               @RequestParam(name = "edns_client_subnet", required = false, defaultValue = "") String ednsClientSubnet,
                               @RequestParam(name = "random_padding", required = false, defaultValue = "") String randomPadding
) {
    String appName = name.endsWith(".") ? name.substring(0, name.length() - 1) : name;
    DnsRecordType dnsRecordType = DnsRecordType.valueOf(type.toUpperCase());
    DnsResponse response = new DnsResponse();
    response.addQuestion(new Question(name, dnsRecordType.intValue()));
    response.setComment("Response from the broker " + RSocketAppContext.ID);
    return dnsResolveService.resolve(appName, type).collectList().map(answers -> {
        response.setAnswers(answers);
        response.setStatus(0);
        return response;
    }).switchIfEmpty(Mono.fromCallable(() -> {
        response.setStatus(-1);
        return response;
    }));
}
 

private static Map<String, String> buildAliasMap(DnsResponse response) {
    final int answerCount = response.count(DnsSection.ANSWER);
    Map<String, String> cnames = null;
    for (int i = 0; i < answerCount; i ++) {
        final DnsRecord r = response.recordAt(DnsSection.ANSWER, i);
        final DnsRecordType type = r.type();
        if (type != DnsRecordType.CNAME) {
            continue;
        }

        if (!(r instanceof DnsRawRecord)) {
            continue;
        }

        final ByteBuf recordContent = ((ByteBufHolder) r).content();
        final String domainName = decodeDomainName(recordContent);
        if (domainName == null) {
            continue;
        }

        if (cnames == null) {
            cnames = new HashMap<String, String>(min(8, answerCount));
        }

        cnames.put(r.name().toLowerCase(Locale.US), domainName.toLowerCase(Locale.US));
    }

    return cnames != null? cnames : Collections.<String, String>emptyMap();
}
 

private static void assertQueryObserver(DnsNameResolver resolver, DnsRecordType cancelledType) {
    TestRecursiveCacheDnsQueryLifecycleObserverFactory lifecycleObserverFactory =
            (TestRecursiveCacheDnsQueryLifecycleObserverFactory) resolver.dnsQueryLifecycleObserverFactory();
    TestDnsQueryLifecycleObserver observer;
    while ((observer = lifecycleObserverFactory.observers.poll()) != null) {
        Object o = observer.events.poll();
        if (o instanceof QueryCancelledEvent) {
            assertEquals(cancelledType, observer.question.type());
        } else if (o instanceof QueryWrittenEvent) {
            QuerySucceededEvent succeededEvent = (QuerySucceededEvent) observer.events.poll();
        } else {
            fail("unexpected event type: " + o);
        }
        assertTrue(observer.events.isEmpty());
    }
}
 

private static ImmutableList<DnsRecordType> dnsRecordTypes(ResolvedAddressTypes resolvedAddressTypes) {
    final Builder<DnsRecordType> builder = ImmutableList.builder();
    switch (resolvedAddressTypes) {
        case IPV4_ONLY:
            builder.add(DnsRecordType.A);
            break;
        case IPV4_PREFERRED:
            builder.add(DnsRecordType.A);
            builder.add(DnsRecordType.AAAA);
            break;
        case IPV6_PREFERRED:
            builder.add(DnsRecordType.AAAA);
            builder.add(DnsRecordType.A);
            break;
    }
    return builder.build();
}
 

@Nullable
public static byte[] extractAddressBytes(DnsRecord record, Logger logger, String logPrefix) {
    final DnsRecordType type = record.type();
    final ByteBuf content = ((ByteBufHolder) record).content();
    final int contentLen = content.readableBytes();

    // Skip invalid records.
    if (type == DnsRecordType.A) {
        if (contentLen != 4) {
            warnInvalidRecord(logger, logPrefix, type, content);
            return null;
        }
    } else if (type == DnsRecordType.AAAA) {
        if (contentLen != 16) {
            warnInvalidRecord(logger, logPrefix, type, content);
            return null;
        }
    } else {
        return null;
    }

    final byte[] addrBytes = new byte[contentLen];
    content.getBytes(content.readerIndex(), addrBytes);
    return addrBytes;
}
 

@Override
public Flux<Answer> resolve(String name, String type) {
    String key = name + ":" + type;
    DnsRecordType dnsRecordType = DnsRecordType.valueOf(type);
    if (dnsRecordStore.containsKey(key)) {
        return Flux.fromIterable(dnsRecordStore.get(key));
    }
    Collection<RSocketBrokerResponderHandler> handlers = handlerRegistry.findByAppName(name);
    if (handlers != null && !handlers.isEmpty()) {
        return Flux.fromIterable(handlers)
                .filter(handler -> handler.getAppStatus().equals(AppStatusEvent.STATUS_SERVING))
                .map(handler -> new Answer(handler.getAppMetadata().getName(), dnsRecordType.intValue(), 300, handler.getAppMetadata().getIp()));
    }
    return Flux.empty();
}
 

@Override
public void addRecords(String name, String type, String... datas) {
    final DnsRecordType recordType = DnsRecordType.valueOf(type);
    addRecords(name, type,
            Stream.of(datas)
                    .map(data -> new Answer(name, recordType.intValue(), 300, data))
                    .collect(Collectors.toList()));

}
 

void onResponse(final DnsServerAddressStream nameServerAddrStream, final int nameServerAddrStreamIndex,
                final DnsQuestion question, AddressedEnvelope<DnsResponse, InetSocketAddress> envelope,
                final DnsQueryLifecycleObserver queryLifecycleObserver,
                Promise<T> promise) {
    try {
        final DnsResponse res = envelope.content();
        final DnsResponseCode code = res.code();
        if (code == DnsResponseCode.NOERROR) {
            if (handleRedirect(question, envelope, queryLifecycleObserver, promise)) {
                // Was a redirect so return here as everything else is handled in handleRedirect(...)
                return;
            }
            final DnsRecordType type = question.type();

            if (type == DnsRecordType.A || type == DnsRecordType.AAAA) {
                onResponseAorAAAA(type, question, envelope, queryLifecycleObserver, promise);
            } else if (type == DnsRecordType.CNAME) {
                onResponseCNAME(question, envelope, queryLifecycleObserver, promise);
            } else {
                queryLifecycleObserver.queryFailed(UNRECOGNIZED_TYPE_QUERY_FAILED_EXCEPTION);
            }
            return;
        }

        // Retry with the next server if the server did not tell us that the domain does not exist.
        if (code != DnsResponseCode.NXDOMAIN) {
            query(nameServerAddrStream, nameServerAddrStreamIndex + 1, question,
                  queryLifecycleObserver.queryNoAnswer(code), promise, null);
        } else {
            queryLifecycleObserver.queryFailed(NXDOMAIN_QUERY_FAILED_EXCEPTION);
        }
    } finally {
        ReferenceCountUtil.safeRelease(envelope);
    }
}
 

private boolean query(String hostname, DnsRecordType type, DnsServerAddressStream dnsServerAddressStream,
                      Promise<T> promise, Throwable cause) {
    final DnsQuestion question = newQuestion(hostname, type);
    if (question == null) {
        return false;
    }
    query(dnsServerAddressStream, 0, question, promise, cause);
    return true;
}
 

private static DnsQuestion newQuestion(String hostname, DnsRecordType type) {
    try {
        return new DefaultDnsQuestion(hostname, type);
    } catch (IllegalArgumentException e) {
        // java.net.IDN.toASCII(...) may throw an IllegalArgumentException if it fails to parse the hostname
        return null;
    }
}
 

RefreshingAddressResolver(EventLoop eventLoop,
                          ConcurrentMap<String, CompletableFuture<CacheEntry>> cache,
                          DefaultDnsNameResolver resolver, List<DnsRecordType> dnsRecordTypes,
                          int minTtl, int maxTtl, int negativeTtl, Backoff refreshBackoff) {
    super(eventLoop);
    this.cache = cache;
    this.resolver = resolver;
    this.dnsRecordTypes = dnsRecordTypes;
    this.minTtl = minTtl;
    this.maxTtl = maxTtl;
    this.negativeTtl = negativeTtl;
    this.refreshBackoff = refreshBackoff;
}
 

DnsTextEndpointGroup(EndpointSelectionStrategy selectionStrategy, EventLoop eventLoop,
                     int minTtl, int maxTtl, long queryTimeoutMillis,
                     DnsServerAddressStreamProvider serverAddressStreamProvider,
                     Backoff backoff, String hostname, Function<byte[], Endpoint> mapping) {
    super(selectionStrategy, eventLoop, minTtl, maxTtl, queryTimeoutMillis, serverAddressStreamProvider,
          backoff, ImmutableList.of(DnsQuestionWithoutTrailingDot.of(hostname, DnsRecordType.TXT)),
          unused -> {});
    this.mapping = mapping;
    start();
}
 

DnsServiceEndpointGroup(EndpointSelectionStrategy selectionStrategy,
                        EventLoop eventLoop, int minTtl, int maxTtl,
                        long queryTimeoutMillis, DnsServerAddressStreamProvider serverAddressStreamProvider,
                        Backoff backoff, String hostname) {
    super(selectionStrategy, eventLoop, minTtl, maxTtl, queryTimeoutMillis, serverAddressStreamProvider,
          backoff, ImmutableList.of(DnsQuestionWithoutTrailingDot.of(hostname, DnsRecordType.SRV)),
          unused -> {});
    start();
}
 

private static List<DnsQuestion> newQuestions(
        String hostname, @Nullable ResolvedAddressTypes resolvedAddressTypes) {

    if (resolvedAddressTypes == null) {
        if (NetUtil.isIpV4StackPreferred() || !anyInterfaceSupportsIpV6()) {
            resolvedAddressTypes = ResolvedAddressTypes.IPV4_ONLY;
        } else {
            resolvedAddressTypes = ResolvedAddressTypes.IPV4_PREFERRED;
        }
    }

    final ImmutableList.Builder<DnsQuestion> builder = ImmutableList.builder();
    switch (resolvedAddressTypes) {
        case IPV4_ONLY:
        case IPV4_PREFERRED:
        case IPV6_PREFERRED:
            builder.add(DnsQuestionWithoutTrailingDot.of(hostname, DnsRecordType.A));
            break;
    }
    switch (resolvedAddressTypes) {
        case IPV6_ONLY:
        case IPV4_PREFERRED:
        case IPV6_PREFERRED:
            builder.add(DnsQuestionWithoutTrailingDot.of(hostname, DnsRecordType.AAAA));
            break;
    }
    return builder.build();
}
 

/**
 * Logs a warning message about an invalid record.
 */
public static void warnInvalidRecord(Logger logger, String logPrefix, DnsRecordType type, ByteBuf content) {
    if (logger.isWarnEnabled()) {
        final String dump = ByteBufUtil.hexDump(content);
        logger.warn("{} Skipping invalid {} record: {}",
                    logPrefix, type.name(), dump.isEmpty() ? "<empty>" : dump);
    }
}
 

public DefaultDnsNameResolver(DnsNameResolver delegate, EventLoop eventLoop, long queryTimeoutMillis) {
    this.delegate = requireNonNull(delegate, "delegate");
    this.eventLoop = requireNonNull(eventLoop, "eventLoop");
    checkArgument(queryTimeoutMillis >= 0, "queryTimeoutMillis: %s (expected: >= 0)", queryTimeoutMillis);
    this.queryTimeoutMillis = queryTimeoutMillis;

    if (delegate.resolvedAddressTypes() == ResolvedAddressTypes.IPV6_PREFERRED) {
        preferredOrder = Ordering.explicit(DnsRecordType.AAAA, DnsRecordType.A);
    } else {
        preferredOrder = Ordering.explicit(DnsRecordType.A, DnsRecordType.AAAA);
    }
}
 

@Test
public void testParseRecords() {
    ByteBuf buf = Unpooled.buffer();
    buf.writeShort(0); // priority
    buf.writeShort(0); // weight
    buf.writeShort(993); // port
    encodeName("localhost.", buf);
    DnsRecord record = new DefaultDnsRawRecord("_myprotocol._tcp.crate.io.", DnsRecordType.SRV, 30, buf);

    List<TransportAddress> addresses = srvUnicastHostsProvider.parseRecords(Collections.singletonList(record));
    assertThat(addresses.get(0).getAddress(), is("127.0.0.1"));
    assertThat(addresses.get(0).getPort(), is(993));
}
 

@SuppressWarnings("ConstantConditions")
public List<Tuple2<String, String>> domains() {
    return resolveService.allDomains()
            .map(name -> Tuples.of(name, String.join(", ", resolveService.resolve(name, DnsRecordType.A.name()).map(Answer::getData).collectList().block())))
            .collectList().block();
}
 

public DnsResolveServiceImpl() {
    addRecords("www.taobao.com", DnsRecordType.A.name(), "47.246.24.234", "47.246.25.233");
}
 

/**
 * Handles a redirect answer if needed and returns {@code true} if a redirect query has been made.
 */
private boolean handleRedirect(
        DnsQuestion question, AddressedEnvelope<DnsResponse, InetSocketAddress> envelope,
        final DnsQueryLifecycleObserver queryLifecycleObserver, Promise<T> promise) {
    final DnsResponse res = envelope.content();

    // Check if we have answers, if not this may be an non authority NS and so redirects must be handled.
    if (res.count(DnsSection.ANSWER) == 0) {
        AuthoritativeNameServerList serverNames = extractAuthoritativeNameServers(question.name(), res);

        if (serverNames != null) {
            List<InetSocketAddress> nameServers = new ArrayList<InetSocketAddress>(serverNames.size());
            int additionalCount = res.count(DnsSection.ADDITIONAL);

            for (int i = 0; i < additionalCount; i++) {
                final DnsRecord r = res.recordAt(DnsSection.ADDITIONAL, i);

                if (r.type() == DnsRecordType.A && !parent.supportsARecords() ||
                    r.type() == DnsRecordType.AAAA && !parent.supportsAAAARecords()) {
                    continue;
                }

                final String recordName = r.name();
                AuthoritativeNameServer authoritativeNameServer =
                        serverNames.remove(recordName);

                if (authoritativeNameServer == null) {
                    // Not a server we are interested in.
                    continue;
                }

                InetAddress resolved = parseAddress(r, recordName);
                if (resolved == null) {
                    // Could not parse it, move to the next.
                    continue;
                }

                nameServers.add(new InetSocketAddress(resolved, parent.dnsRedirectPort(resolved)));
                addNameServerToCache(authoritativeNameServer, resolved, r.timeToLive());
            }

            if (!nameServers.isEmpty()) {
                query(parent.uncachedRedirectDnsServerStream(nameServers), 0, question,
                      queryLifecycleObserver.queryRedirected(unmodifiableList(nameServers)), promise, null);
                return true;
            }
        }
    }
    return false;
}
 

void tryToFinishResolve(final DnsServerAddressStream nameServerAddrStream,
                        final int nameServerAddrStreamIndex,
                        final DnsQuestion question,
                        final DnsQueryLifecycleObserver queryLifecycleObserver,
                        final Promise<T> promise,
                        final Throwable cause) {
    // There are no queries left to try.
    if (!queriesInProgress.isEmpty()) {
        queryLifecycleObserver.queryCancelled(allowedQueries);

        // There are still some queries we did not receive responses for.
        if (gotPreferredAddress()) {
            // But it's OK to finish the resolution process if we got a resolved address of the preferred type.
            finishResolve(promise, cause);
        }

        // We did not get any resolved address of the preferred type, so we can't finish the resolution process.
        return;
    }

    // There are no queries left to try.
    if (resolvedEntries == null) {
        if (nameServerAddrStreamIndex < nameServerAddrStream.size()) {
            if (queryLifecycleObserver == NoopDnsQueryLifecycleObserver.INSTANCE) {
                // If the queryLifecycleObserver has already been terminated we should create a new one for this
                // fresh query.
                query(nameServerAddrStream, nameServerAddrStreamIndex + 1, question, promise, cause);
            } else {
                query(nameServerAddrStream, nameServerAddrStreamIndex + 1, question, queryLifecycleObserver,
                      promise, cause);
            }
            return;
        }

        queryLifecycleObserver.queryFailed(NAME_SERVERS_EXHAUSTED_EXCEPTION);

        // .. and we could not find any A/AAAA records.

        // If cause != null we know this was caused by a timeout / cancel / transport exception. In this case we
        // won't try to resolve the CNAME as we only should do this if we could not get the A/AAAA records because
        // these not exists and the DNS server did probably signal it.
        if (cause == null && !triedCNAME) {
            // As the last resort, try to query CNAME, just in case the name server has it.
            triedCNAME = true;

            query(hostname, DnsRecordType.CNAME, getNameServers(hostname), promise, null);
            return;
        }
    } else {
        queryLifecycleObserver.queryCancelled(allowedQueries);
    }

    // We have at least one resolved address or tried CNAME as the last resort..
    finishResolve(promise, cause);
}
 

void add(DnsRecord r) {
    if (r.type() != DnsRecordType.NS || !(r instanceof DnsRawRecord)) {
        return;
    }

    // Only include servers that serve the correct domain.
    if (questionName.length() <  r.name().length()) {
        return;
    }

    String recordName = r.name().toLowerCase(Locale.US);

    int dots = 0;
    for (int a = recordName.length() - 1, b = questionName.length() - 1; a >= 0; a--, b--) {
        char c = recordName.charAt(a);
        if (questionName.charAt(b) != c) {
            return;
        }
        if (c == '.') {
            dots++;
        }
    }

    if (head != null && head.dots > dots) {
        // We already have a closer match so ignore this one, no need to parse the domainName etc.
        return;
    }

    final ByteBuf recordContent = ((ByteBufHolder) r).content();
    final String domainName = decodeDomainName(recordContent);
    if (domainName == null) {
        // Could not be parsed, ignore.
        return;
    }

    // We are only interested in preserving the nameservers which are the closest to our qName, so ensure
    // we drop servers that have a smaller dots count.
    if (head == null || head.dots < dots) {
        count = 1;
        head = new AuthoritativeNameServer(dots, recordName, domainName);
    } else if (head.dots == dots) {
        AuthoritativeNameServer serverName = head;
        while (serverName.next != null) {
            serverName = serverName.next;
        }
        serverName.next = new AuthoritativeNameServer(dots, recordName, domainName);
        count++;
    }
}
 

final DnsRecordType[] resolveRecordTypes() {
    return resolveRecordTypes;
}
 

private static Map<String, InetAddress> testResolve0(DnsNameResolver resolver, Set<String> excludedDomains,
                                                     DnsRecordType cancelledType)
        throws InterruptedException {

    assertThat(resolver.isRecursionDesired(), is(true));

    final Map<String, InetAddress> results = new HashMap<String, InetAddress>();
    final Map<String, Future<InetAddress>> futures =
            new LinkedHashMap<String, Future<InetAddress>>();

    for (String name : DOMAINS) {
        if (excludedDomains.contains(name)) {
            continue;
        }

        resolve(resolver, futures, name);
    }

    for (Entry<String, Future<InetAddress>> e : futures.entrySet()) {
        String unresolved = e.getKey();
        InetAddress resolved = e.getValue().sync().getNow();

        logger.info("{}: {}", unresolved, resolved.getHostAddress());

        assertThat(resolved.getHostName(), is(unresolved));

        boolean typeMatches = false;
        for (InternetProtocolFamily f: resolver.resolvedInternetProtocolFamiliesUnsafe()) {
            Class<?> resolvedType = resolved.getClass();
            if (f.addressType().isAssignableFrom(resolvedType)) {
                typeMatches = true;
            }
        }

        assertThat(typeMatches, is(true));

        results.put(resolved.getHostName(), resolved);
    }

    assertQueryObserver(resolver, cancelledType);

    return results;
}
 

private static void queryMx(
        DnsNameResolver resolver,
        Map<String, Future<AddressedEnvelope<DnsResponse, InetSocketAddress>>> futures,
        String hostname) throws Exception {
    futures.put(hostname, resolver.query(new DefaultDnsQuestion(hostname, DnsRecordType.MX)));
}
 

@Override
ImmutableSortedSet<Endpoint> onDnsRecords(List<DnsRecord> records, int ttl) throws Exception {
    final ImmutableSortedSet.Builder<Endpoint> builder = ImmutableSortedSet.naturalOrder();
    for (DnsRecord r : records) {
        if (!(r instanceof DnsRawRecord) || r.type() != DnsRecordType.TXT) {
            continue;
        }

        final ByteBuf content = ((ByteBufHolder) r).content();
        if (!content.isReadable()) { // Missing length octet
            warnInvalidRecord(DnsRecordType.TXT, content);
            continue;
        }

        content.markReaderIndex();
        final int txtLen = content.readUnsignedByte();
        if (txtLen == 0) { // Empty content
            continue;
        }

        if (content.readableBytes() != txtLen) { // Mismatching number of octets
            content.resetReaderIndex();
            warnInvalidRecord(DnsRecordType.TXT, content);
            continue;
        }

        final byte[] txt = new byte[txtLen];
        content.readBytes(txt);

        final Endpoint endpoint;
        try {
            endpoint = mapping.apply(txt);
        } catch (Exception e) {
            content.resetReaderIndex();
            warnInvalidRecord(DnsRecordType.TXT, content);
            continue;
        }

        if (endpoint != null) {
            builder.add(endpoint);
        }
    }

    final ImmutableSortedSet<Endpoint> endpoints = builder.build();
    if (logger().isDebugEnabled()) {
        logger().debug("{} Resolved: {} (TTL: {})",
                       logPrefix(),
                       endpoints.stream().map(Object::toString).collect(Collectors.joining(", ")),
                       ttl);
    }

    return endpoints;
}
 

@Override
ImmutableSortedSet<Endpoint> onDnsRecords(List<DnsRecord> records, int ttl) throws Exception {
    final ImmutableSortedSet.Builder<Endpoint> builder = ImmutableSortedSet.naturalOrder();
    for (DnsRecord r : records) {
        if (!(r instanceof DnsRawRecord) || r.type() != DnsRecordType.SRV) {
            continue;
        }

        final ByteBuf content = ((ByteBufHolder) r).content();
        if (content.readableBytes() <= 6) { // Too few bytes
            warnInvalidRecord(DnsRecordType.SRV, content);
            continue;
        }

        content.markReaderIndex();
        content.skipBytes(2);  // priority unused
        final int weight = content.readUnsignedShort();
        final int port = content.readUnsignedShort();

        final Endpoint endpoint;
        try {
            final String target = stripTrailingDot(DefaultDnsRecordDecoder.decodeName(content));
            endpoint = port > 0 ? Endpoint.of(target, port) : Endpoint.of(target);
        } catch (Exception e) {
            content.resetReaderIndex();
            warnInvalidRecord(DnsRecordType.SRV, content);
            continue;
        }

        builder.add(endpoint.withWeight(weight));
    }

    final ImmutableSortedSet<Endpoint> endpoints = builder.build();
    if (logger().isDebugEnabled()) {
        logger().debug("{} Resolved: {} (TTL: {})",
                       logPrefix(),
                       endpoints.stream()
                                .map(e -> e.authority() + '/' + e.weight())
                                .collect(Collectors.joining(", ")),
                       ttl);
    }

    return endpoints;
}
 

/**
 * Logs a warning message about an invalid record.
 */
final void warnInvalidRecord(DnsRecordType type, ByteBuf content) {
    DnsUtil.warnInvalidRecord(logger(), logPrefix(), type, content);
}
 

public static DnsQuestionWithoutTrailingDot of(String name, DnsRecordType type) {
    return new DnsQuestionWithoutTrailingDot(name, type);
}
 

private DnsQuestionWithoutTrailingDot(String name, DnsRecordType type) {
    this.name = IDN.toASCII(requireNonNull(name, "name"));
    this.type = requireNonNull(type, "type");
}