io.netty.channel.AddressedEnvelope#content ( )源码实例Demo

@Override
protected void encode(ChannelHandlerContext ctx,
                      AddressedEnvelope<DnsResponse, InetSocketAddress> in, List<Object> out) throws Exception {

    final InetSocketAddress recipient = in.recipient();
    final DnsResponse response = in.content();
    final ByteBuf buf = allocateBuffer(ctx, in);

    boolean success = false;
    try {
        encodeHeader(response, buf);
        encodeQuestions(response, buf);
        encodeRecords(response, DnsSection.ANSWER, buf);
        encodeRecords(response, DnsSection.AUTHORITY, buf);
        encodeRecords(response, DnsSection.ADDITIONAL, buf);
        success = true;
    } finally {
        if (!success) {
            buf.release();
        }
    }

    out.add(new DatagramPacket(buf, recipient, null));
}
 

@Override
protected void encode(
    ChannelHandlerContext ctx,
    AddressedEnvelope<DnsQuery, InetSocketAddress> in, List<Object> out) throws Exception {

    final InetSocketAddress recipient = in.recipient();
    final DnsQuery query = in.content();
    final ByteBuf buf = allocateBuffer(ctx, in);

    boolean success = false;
    try {
        encodeHeader(query, buf);
        encodeQuestions(query, buf);
        encodeRecords(query, DnsSection.ADDITIONAL, buf);
        success = true;
    } finally {
        if (!success) {
            buf.release();
        }
    }

    out.add(new DatagramPacket(buf, recipient, null));
}
 

@Test
public void readResponseTest() throws Exception {
    EmbeddedChannel embedder = new EmbeddedChannel(new DatagramDnsResponseDecoder());
    for (byte[] p: packets) {
        ByteBuf packet = embedder.alloc().buffer(512).writeBytes(p);
        embedder.writeInbound(new DatagramPacket(packet, null, new InetSocketAddress(0)));
        AddressedEnvelope<DnsResponse, InetSocketAddress> envelope = embedder.readInbound();
        assertThat(envelope, is(instanceOf(DatagramDnsResponse.class)));
        DnsResponse response = envelope.content();
        assertThat(response, is(sameInstance((Object) envelope)));

        ByteBuf raw = Unpooled.wrappedBuffer(p);
        assertThat(response.id(), is(raw.getUnsignedShort(0)));
        assertThat(response.count(DnsSection.QUESTION), is(raw.getUnsignedShort(4)));
        assertThat(response.count(DnsSection.ANSWER), is(raw.getUnsignedShort(6)));
        assertThat(response.count(DnsSection.AUTHORITY), is(raw.getUnsignedShort(8)));
        assertThat(response.count(DnsSection.ADDITIONAL), is(raw.getUnsignedShort(10)));

        envelope.release();
    }
}
 

@Override
protected Object filterOutboundMessage(Object msg) {
    if (msg instanceof DatagramPacket || msg instanceof ByteBuf) {
        return msg;
    }

    if (msg instanceof AddressedEnvelope) {
        @SuppressWarnings("unchecked")
        AddressedEnvelope<Object, SocketAddress> e = (AddressedEnvelope<Object, SocketAddress>) msg;
        if (e.content() instanceof ByteBuf) {
            return msg;
        }
    }

    throw new UnsupportedOperationException(
            "unsupported message type: " + StringUtil.simpleClassName(msg) + EXPECTED_TYPES);
}
 

@Override
protected Object filterOutboundMessage(Object msg) {
    if (msg instanceof DatagramPacket || msg instanceof ByteBuf) {
        return msg;
    }

    if (msg instanceof AddressedEnvelope) {
        @SuppressWarnings("unchecked")
        AddressedEnvelope<Object, SocketAddress> e = (AddressedEnvelope<Object, SocketAddress>) msg;
        if (e.content() instanceof ByteBuf) {
            return msg;
        }
    }

    throw new UnsupportedOperationException(
            "unsupported message type: " + StringUtil.simpleClassName(msg) + EXPECTED_TYPES);
}
 

void finish(AddressedEnvelope<? extends DnsResponse, InetSocketAddress> envelope) {
    final DnsResponse res = envelope.content();
    if (res.count(DnsSection.QUESTION) != 1) {
        logger.warn("Received a DNS response with invalid number of questions: {}", envelope);
        return;
    }

    if (!question().equals(res.recordAt(DnsSection.QUESTION))) {
        logger.warn("Received a mismatching DNS response: {}", envelope);
        return;
    }

    setSuccess(envelope);
}
 

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);
    }
}
 

/**
 * 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;
}
 

private void resolveAddress() {
    boolean debug = getFlag(BuiltinFlags.PRINT_DEBUG, false);

    String name = this.getPacketProtocol().getSRVRecordPrefix() + "._tcp." + this.getHost();
    if(debug) {
        System.out.println("[PacketLib] Attempting SRV lookup for \"" + name + "\".");
    }

    AddressedEnvelope<DnsResponse, InetSocketAddress> envelope = null;
    try(DnsNameResolver resolver = new DnsNameResolverBuilder(this.group.next())
            .channelType(NioDatagramChannel.class)
            .build()) {
        envelope = resolver.query(new DefaultDnsQuestion(name, DnsRecordType.SRV)).get();
        DnsResponse response = envelope.content();
        if(response.count(DnsSection.ANSWER) > 0) {
            DefaultDnsRawRecord record = response.recordAt(DnsSection.ANSWER, 0);
            if(record.type() == DnsRecordType.SRV) {
                ByteBuf buf = record.content();
                buf.skipBytes(4); // Skip priority and weight.

                int port = buf.readUnsignedShort();
                String host = DefaultDnsRecordDecoder.decodeName(buf);
                if(host.endsWith(".")) {
                    host = host.substring(0, host.length() - 1);
                }

                if(debug) {
                    System.out.println("[PacketLib] Found SRV record containing \"" + host + ":" + port + "\".");
                }

                this.host = host;
                this.port = port;
            } else if(debug) {
                System.out.println("[PacketLib] Received non-SRV record in response.");
            }
        } else if(debug) {
            System.out.println("[PacketLib] No SRV record found.");
        }
    } catch(Exception e) {
        if(debug) {
            System.out.println("[PacketLib] Failed to resolve SRV record.");
            e.printStackTrace();
        }
    } finally {
        if(envelope != null) {
            envelope.release();
        }
    }
}