Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
- Loading branch information
Showing
3 changed files
with
153 additions
and
13 deletions.
There are no files selected for viewing
59 changes: 59 additions & 0 deletions
59
akka-actor-tests/src/test/scala/akka/io/dns/OnlineAsyncDnsResolverIntegrationSpec.scala
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,59 @@ | ||
| /* | ||
| * Copyright (C) 2018 Lightbend Inc. <https://www.lightbend.com> | ||
| */ | ||
|
|
||
| package akka.io.dns | ||
|
|
||
| import java.net.InetAddress | ||
|
|
||
| import akka.io.dns.DnsProtocol.{ Ip, RequestType, Srv } | ||
| import akka.io.{ Dns, IO } | ||
| import akka.pattern.ask | ||
| import akka.testkit.AkkaSpec | ||
| import akka.util.Timeout | ||
|
|
||
| import scala.concurrent.duration._ | ||
|
|
||
| /* | ||
| Relies on being run while online | ||
| */ | ||
| class OnlineAsyncDnsResolverIntegrationSpec extends AkkaSpec( | ||
| """ | ||
| akka.loglevel = DEBUG | ||
| akka.io.dns.resolver = async-dns | ||
| akka.io.dns.async-dns.nameservers = default | ||
| """) { | ||
| val duration = 10.seconds | ||
| implicit val timeout = Timeout(duration) | ||
|
|
||
| "Resolver" must { | ||
|
|
||
| "resolve mixed A/AAAA records" in { | ||
| val name = "akka.io" | ||
| val answer = resolve(name) | ||
| answer.name shouldEqual name | ||
|
|
||
| answer.records.collect { case r: ARecord ⇒ r.ip }.toSet shouldEqual Set( | ||
| InetAddress.getByName("104.31.90.133"), | ||
| InetAddress.getByName("104.31.91.133") | ||
| ) | ||
|
|
||
| answer.records.collect { case r: AAAARecord ⇒ r.ip }.toSet shouldEqual Set( | ||
| InetAddress.getByName("2606:4700:30::681f:5a85"), | ||
| InetAddress.getByName("2606:4700:30::681f:5b85") | ||
| ) | ||
| } | ||
|
|
||
| "resolve queries that are too big for UDP" in { | ||
| val name = "many.bzzt.net" | ||
| val answer = resolve(name) | ||
| answer.name shouldEqual name | ||
| answer.records.length should be(48) | ||
| } | ||
|
|
||
| def resolve(name: String, requestType: RequestType = Ip()): DnsProtocol.Resolved = { | ||
| (IO(Dns) ? DnsProtocol.Resolve(name, requestType)).mapTo[DnsProtocol.Resolved].futureValue | ||
| } | ||
|
|
||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
71 changes: 71 additions & 0 deletions
71
akka-actor/src/main/scala/akka/io/dns/internal/TcpDnsClient.scala
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,71 @@ | ||
| /* | ||
| * Copyright (C) 2018 Lightbend Inc. <https://www.lightbend.com> | ||
| */ | ||
|
|
||
| package akka.io.dns.internal | ||
|
|
||
| import java.net.InetSocketAddress | ||
|
|
||
| import akka.actor.{ Actor, ActorLogging, ActorRef, Stash } | ||
| import akka.annotation.InternalApi | ||
| import akka.io.Tcp._ | ||
| import akka.io.dns.internal.DnsClient.{ Answer, DnsQuestion, Question4 } | ||
| import akka.io.{ IO, Tcp } | ||
| import akka.util.ByteString | ||
|
|
||
| /** | ||
| * INTERNAL API | ||
| */ | ||
| @InternalApi private[akka] class TcpDnsClient(ns: InetSocketAddress) extends Actor with ActorLogging with Stash { | ||
|
|
||
| import context.system | ||
|
|
||
| log.warning("Connecting to [{}]", ns) | ||
| IO(Tcp) ! Tcp.Connect(ns) | ||
|
|
||
| override def receive: Receive = { | ||
| case CommandFailed(_: Connect) ⇒ | ||
| log.warning("Failed to connect to [{}]", ns) | ||
| // TODO | ||
| case _: Tcp.Connected ⇒ | ||
| log.debug(s"Connected to TCP address [{}]", ns) | ||
| val connection = sender() | ||
| context.become(ready(connection)) | ||
| connection ! Register(self) | ||
| unstashAll() | ||
| case _: Message ⇒ | ||
| stash() | ||
| } | ||
|
|
||
| def encodeLength(length: Int): ByteString = | ||
| ByteString((length / 256).toByte, length.toByte) | ||
|
|
||
| def decodeLength(data: ByteString): Int = | ||
| ((data(0).toInt + 256) % 256) * 256 + ((data(1) + 256) % 256) | ||
|
|
||
| def ready(connection: ActorRef): Receive = { | ||
| case msg: Message ⇒ | ||
| log.warning("Sending message to connection") | ||
| val bytes = msg.write() | ||
| connection ! Tcp.Write(encodeLength(bytes.length)) | ||
| connection ! Tcp.Write(bytes) | ||
| case CommandFailed(_: Write) ⇒ | ||
| // TODO | ||
| log.warning("Write failed") | ||
| case Received(data) ⇒ | ||
| log.warning("Received data") | ||
| require(data.length > 2, "Expected a response datagram starting with the size") | ||
| val expectedLength = decodeLength(data) | ||
| log.warning(s"First 2 bytes are ${data(0)} and ${data(1)}, totalling $expectedLength") | ||
| require(data.length == expectedLength + 2, s"Expected a full response datagram of length ${expectedLength}, got ${data.length - 2} data bytes instead.") | ||
| val msg = Message.parse(data.drop(2)) | ||
| log.debug(s"Decoded: $msg") | ||
| if (msg.flags.isTruncated) { | ||
| log.warning("TCP DNS response truncated") | ||
| } | ||
| val (recs, additionalRecs) = if (msg.flags.responseCode == ResponseCode.SUCCESS) (msg.answerRecs, msg.additionalRecs) else (Nil, Nil) | ||
| context.parent ! Answer(msg.id, recs, additionalRecs) | ||
| case PeerClosed ⇒ | ||
| log.warning("Peer closed") | ||
| } | ||
| } |