下面列出了 io.netty.handler.codec.dns.DnsSection #io.netty.channel.AddressedEnvelope 实例代码,或者点击链接到github查看源代码,也可以在右侧发表评论。
private static StringBuilder appendAddresses(StringBuilder buf, DnsMessage msg) {
if (!(msg instanceof AddressedEnvelope)) {
return buf;
}
@SuppressWarnings("unchecked")
AddressedEnvelope<?, SocketAddress> envelope = (AddressedEnvelope<?, SocketAddress>) msg;
SocketAddress addr = envelope.sender();
if (addr != null) {
buf.append("from: ")
.append(addr)
.append(", ");
}
addr = envelope.recipient();
if (addr != null) {
buf.append("to: ")
.append(addr)
.append(", ");
}
return buf;
}
@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();
}
}
DnsQueryContext(DnsNameResolver parent,
InetSocketAddress nameServerAddr,
DnsQuestion question,
DnsRecord[] additionals,
Promise<AddressedEnvelope<DnsResponse, InetSocketAddress>> promise) {
this.parent = checkNotNull(parent, "parent");
this.nameServerAddr = checkNotNull(nameServerAddr, "nameServerAddr");
this.question = checkNotNull(question, "question");
this.additionals = checkNotNull(additionals, "additionals");
this.promise = checkNotNull(promise, "promise");
recursionDesired = parent.isRecursionDesired();
id = parent.queryContextManager.add(this);
if (parent.isOptResourceEnabled()) {
optResource = new AbstractDnsOptPseudoRrRecord(parent.maxPayloadSize(), 0, 0) {
// We may want to remove this in the future and let the user just specify the opt record in the query.
};
} else {
optResource = null;
}
}
private void setSuccess(AddressedEnvelope<? extends DnsResponse, InetSocketAddress> envelope) {
parent.queryContextManager.remove(nameServerAddr(), id);
// Cancel the timeout task.
final ScheduledFuture<?> timeoutFuture = this.timeoutFuture;
if (timeoutFuture != null) {
timeoutFuture.cancel(false);
}
Promise<AddressedEnvelope<DnsResponse, InetSocketAddress>> promise = this.promise;
if (promise.setUncancellable()) {
@SuppressWarnings("unchecked")
AddressedEnvelope<DnsResponse, InetSocketAddress> castResponse =
(AddressedEnvelope<DnsResponse, InetSocketAddress>) envelope.retain();
if (!promise.trySuccess(castResponse)) {
// We failed to notify the promise as it was failed before, thus we need to release the envelope
envelope.release();
}
}
}
final Future<AddressedEnvelope<DnsResponse, InetSocketAddress>> query0(
InetSocketAddress nameServerAddr, DnsQuestion question,
DnsRecord[] additionals,
ChannelPromise writePromise,
Promise<AddressedEnvelope<? extends DnsResponse, InetSocketAddress>> promise) {
assert !writePromise.isVoid();
final Promise<AddressedEnvelope<DnsResponse, InetSocketAddress>> castPromise = cast(
checkNotNull(promise, "promise"));
try {
new DnsQueryContext(this, nameServerAddr, question, additionals, castPromise).query(writePromise);
return castPromise;
} catch (Exception e) {
return castPromise.setFailure(e);
}
}
@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 void encode(
ChannelHandlerContext ctx, AddressedEnvelope<M, InetSocketAddress> msg, List<Object> out) throws Exception {
assert out.isEmpty();
encoder.encode(ctx, msg.content(), out);
if (out.size() != 1) {
throw new EncoderException(
StringUtil.simpleClassName(encoder) + " must produce only one message.");
}
Object content = out.get(0);
if (content instanceof ByteBuf) {
// Replace the ByteBuf with a DatagramPacket.
out.set(0, new DatagramPacket((ByteBuf) content, msg.recipient(), msg.sender()));
} else {
throw new EncoderException(
StringUtil.simpleClassName(encoder) + " must produce only ByteBuf.");
}
}
@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
public boolean equals(Object obj) {
if (this == obj) {
return true;
}
if (!super.equals(obj)) {
return false;
}
if (!(obj instanceof AddressedEnvelope)) {
return false;
}
@SuppressWarnings("unchecked")
final AddressedEnvelope<?, SocketAddress> that = (AddressedEnvelope<?, SocketAddress>) obj;
if (sender() == null) {
if (that.sender() != null) {
return false;
}
} else if (!sender().equals(that.sender())) {
return false;
}
if (recipient() == null) {
if (that.recipient() != null) {
return false;
}
} else if (!recipient().equals(that.recipient())) {
return false;
}
return true;
}
@Override
public boolean equals(Object obj) {
if (this == obj) {
return true;
}
if (!super.equals(obj)) {
return false;
}
if (!(obj instanceof AddressedEnvelope)) {
return false;
}
@SuppressWarnings("unchecked")
final AddressedEnvelope<?, SocketAddress> that = (AddressedEnvelope<?, SocketAddress>) obj;
if (sender() == null) {
if (that.sender() != null) {
return false;
}
} else if (!sender().equals(that.sender())) {
return false;
}
if (recipient() == null) {
if (that.recipient() != null) {
return false;
}
} else if (!recipient().equals(that.recipient())) {
return false;
}
return true;
}
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);
}
}
private void onResponseCNAME(
DnsQuestion question, AddressedEnvelope<DnsResponse, InetSocketAddress> response,
Map<String, String> cnames, final DnsQueryLifecycleObserver queryLifecycleObserver,
Promise<T> promise) {
// Resolve the host name in the question into the real host name.
final String name = question.name().toLowerCase(Locale.US);
String resolved = name;
boolean found = false;
while (!cnames.isEmpty()) { // Do not attempt to call Map.remove() when the Map is empty
// because it can be Collections.emptyMap()
// whose remove() throws a UnsupportedOperationException.
final String next = cnames.remove(resolved);
if (next != null) {
found = true;
resolved = next;
} else {
break;
}
}
if (found) {
followCname(question, resolved, queryLifecycleObserver, promise);
} else {
queryLifecycleObserver.queryFailed(CNAME_NOT_FOUND_QUERY_FAILED_EXCEPTION);
}
}
/**
* Sends a DNS query with the specified question using the specified name server list.
*/
public Future<AddressedEnvelope<DnsResponse, InetSocketAddress>> query(
InetSocketAddress nameServerAddr, DnsQuestion question) {
return query0(nameServerAddr, question, EMPTY_ADDITIONALS,
ch.eventLoop().<AddressedEnvelope<? extends DnsResponse, InetSocketAddress>>newPromise());
}
/**
* Sends a DNS query with the specified question with additional records using the specified name server list.
*/
public Future<AddressedEnvelope<DnsResponse, InetSocketAddress>> query(
InetSocketAddress nameServerAddr, DnsQuestion question, Iterable<DnsRecord> additionals) {
return query0(nameServerAddr, question, toArray(additionals, false),
ch.eventLoop().<AddressedEnvelope<? extends DnsResponse, InetSocketAddress>>newPromise());
}
/**
* Sends a DNS query with the specified question using the specified name server list.
*/
public Future<AddressedEnvelope<DnsResponse, InetSocketAddress>> query(
InetSocketAddress nameServerAddr, DnsQuestion question,
Promise<AddressedEnvelope<? extends DnsResponse, InetSocketAddress>> promise) {
return query0(nameServerAddr, question, EMPTY_ADDITIONALS, promise);
}
/**
* Sends a DNS query with the specified question with additional records using the specified name server list.
*/
public Future<AddressedEnvelope<DnsResponse, InetSocketAddress>> query(
InetSocketAddress nameServerAddr, DnsQuestion question,
Iterable<DnsRecord> additionals,
Promise<AddressedEnvelope<? extends DnsResponse, InetSocketAddress>> promise) {
return query0(nameServerAddr, question, toArray(additionals, false), promise);
}
@Override
public boolean acceptOutboundMessage(Object msg) throws Exception {
if (super.acceptOutboundMessage(msg)) {
@SuppressWarnings("rawtypes")
AddressedEnvelope envelope = (AddressedEnvelope) msg;
return encoder.acceptOutboundMessage(envelope.content())
&& envelope.sender() instanceof InetSocketAddress
&& envelope.recipient() instanceof InetSocketAddress;
}
return false;
}
private static void testSharable(boolean sharable) {
MessageToMessageEncoder<AddressedEnvelope<ByteBuf, InetSocketAddress>> wrapped =
new TestMessageToMessageEncoder(sharable);
DatagramPacketEncoder<AddressedEnvelope<ByteBuf, InetSocketAddress>> encoder =
new DatagramPacketEncoder<AddressedEnvelope<ByteBuf, InetSocketAddress>>(wrapped);
assertEquals(wrapped.isSharable(), encoder.isSharable());
}
/**
* Allocate a {@link ByteBuf} which will be used for constructing a datagram packet.
* Sub-classes may override this method to return a {@link ByteBuf} with a perfect matching initial capacity.
*/
protected ByteBuf allocateBuffer(
ChannelHandlerContext ctx,
@SuppressWarnings("unused") AddressedEnvelope<DnsResponse, InetSocketAddress> msg) throws Exception {
return ctx.alloc().ioBuffer(1024);
}
/**
* Allocate a {@link ByteBuf} which will be used for constructing a datagram packet.
* Sub-classes may override this method to return a {@link ByteBuf} with a perfect matching initial capacity.
*/
protected ByteBuf allocateBuffer(
ChannelHandlerContext ctx,
@SuppressWarnings("unused") AddressedEnvelope<DnsQuery, InetSocketAddress> msg) throws Exception {
return ctx.alloc().ioBuffer(1024);
}
@Override
public void operationComplete(Future<AddressedEnvelope<DnsResponse, InetSocketAddress>> future) {
if (future.isSuccess()) {
future.getNow().release();
}
}
private void query(final DnsServerAddressStream nameServerAddrStream,
final int nameServerAddrStreamIndex,
final DnsQuestion question,
final DnsQueryLifecycleObserver queryLifecycleObserver,
final Promise<T> promise,
final Throwable cause) {
if (nameServerAddrStreamIndex >= nameServerAddrStream.size() || allowedQueries == 0 || promise.isCancelled()) {
tryToFinishResolve(nameServerAddrStream, nameServerAddrStreamIndex, question, queryLifecycleObserver,
promise, cause);
return;
}
--allowedQueries;
final InetSocketAddress nameServerAddr = nameServerAddrStream.next();
final ChannelPromise writePromise = parent.ch.newPromise();
final Future<AddressedEnvelope<DnsResponse, InetSocketAddress>> f = parent.query0(
nameServerAddr, question, additionals, writePromise,
parent.ch.eventLoop().<AddressedEnvelope<? extends DnsResponse, InetSocketAddress>>newPromise());
queriesInProgress.add(f);
queryLifecycleObserver.queryWritten(nameServerAddr, writePromise);
f.addListener(new FutureListener<AddressedEnvelope<DnsResponse, InetSocketAddress>>() {
@Override
public void operationComplete(Future<AddressedEnvelope<DnsResponse, InetSocketAddress>> future) {
queriesInProgress.remove(future);
if (promise.isDone() || future.isCancelled()) {
queryLifecycleObserver.queryCancelled(allowedQueries);
// Check if we need to release the envelope itself. If the query was cancelled the getNow() will
// return null as well as the Future will be failed with a CancellationException.
AddressedEnvelope<DnsResponse, InetSocketAddress> result = future.getNow();
if (result != null) {
result.release();
}
return;
}
final Throwable queryCause = future.cause();
try {
if (queryCause == null) {
onResponse(nameServerAddrStream, nameServerAddrStreamIndex, question, future.getNow(),
queryLifecycleObserver, promise);
} else {
// Server did not respond or I/O error occurred; try again.
queryLifecycleObserver.queryFailed(queryCause);
query(nameServerAddrStream, nameServerAddrStreamIndex + 1, question, promise, queryCause);
}
} finally {
tryToFinishResolve(nameServerAddrStream, nameServerAddrStreamIndex, question,
// queryLifecycleObserver has already been terminated at this point so we must
// not allow it to be terminated again by tryToFinishResolve.
NoopDnsQueryLifecycleObserver.INSTANCE,
promise, queryCause);
}
}
});
}
/**
* 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 onResponseCNAME(DnsQuestion question, AddressedEnvelope<DnsResponse, InetSocketAddress> envelope,
final DnsQueryLifecycleObserver queryLifecycleObserver,
Promise<T> promise) {
onResponseCNAME(question, envelope, buildAliasMap(envelope.content()), queryLifecycleObserver, promise);
}
/**
* Sends a DNS query with the specified question.
*/
public Future<AddressedEnvelope<DnsResponse, InetSocketAddress>> query(DnsQuestion question) {
return query(nextNameServerAddress(), question);
}
/**
* Sends a DNS query with the specified question with additional records.
*/
public Future<AddressedEnvelope<DnsResponse, InetSocketAddress>> query(
DnsQuestion question, Iterable<DnsRecord> additionals) {
return query(nextNameServerAddress(), question, additionals);
}
/**
* Sends a DNS query with the specified question.
*/
public Future<AddressedEnvelope<DnsResponse, InetSocketAddress>> query(
DnsQuestion question, Promise<AddressedEnvelope<? extends DnsResponse, InetSocketAddress>> promise) {
return query(nextNameServerAddress(), question, Collections.<DnsRecord>emptyList(), promise);
}