1
0
mirror of https://github.com/danog/dns.git synced 2024-12-12 17:37:21 +01:00
dns/lib/Internal/TcpSocket.php

125 lines
3.2 KiB
PHP
Raw Normal View History

2017-06-23 07:34:11 +02:00
<?php
namespace Amp\Dns\Internal;
2017-06-23 07:34:11 +02:00
use Amp;
2017-06-23 07:34:11 +02:00
use Amp\Deferred;
use Amp\Dns\DnsException;
use Amp\Dns\TimeoutException;
2017-06-23 07:34:11 +02:00
use Amp\Loop;
use Amp\Parser\Parser;
use Amp\Promise;
use Amp\Success;
use LibDNS\Decoder\DecoderFactory;
use LibDNS\Encoder\EncoderFactory;
use LibDNS\Messages\Message;
use function Amp\call;
/** @internal */
2019-01-25 02:27:47 +01:00
class TcpSocket extends Socket
{
/** @var \LibDNS\Encoder\Encoder */
2017-06-23 07:34:11 +02:00
private $encoder;
/** @var \SplQueue */
2017-06-23 07:34:11 +02:00
private $queue;
/** @var Parser */
2017-06-23 07:34:11 +02:00
private $parser;
2017-06-23 12:49:16 +02:00
/** @var bool */
private $isAlive = true;
2019-01-25 02:27:47 +01:00
public static function connect(string $uri, int $timeout = 5000): Promise
{
2017-06-23 07:34:11 +02:00
if (!$socket = @\stream_socket_client($uri, $errno, $errstr, 0, STREAM_CLIENT_ASYNC_CONNECT)) {
throw new DnsException(\sprintf(
2017-06-23 07:34:11 +02:00
"Connection to %s failed: [Error #%d] %s",
$uri,
$errno,
$errstr
));
}
\stream_set_blocking($socket, false);
return call(function () use ($uri, $socket, $timeout) {
$deferred = new Deferred;
2017-06-23 07:34:11 +02:00
$watcher = Loop::onWritable($socket, static function () use ($socket, $deferred) {
$deferred->resolve(new self($socket));
});
try {
return yield Promise\timeout($deferred->promise(), $timeout);
} catch (Amp\TimeoutException $e) {
throw new TimeoutException("Name resolution timed out, could not connect to server at $uri");
} finally {
Loop::cancel($watcher);
}
});
2017-06-23 07:34:11 +02:00
}
2019-01-25 02:27:47 +01:00
public static function parser(callable $callback): \Generator
{
2017-06-23 07:34:11 +02:00
$decoder = (new DecoderFactory)->create();
while (true) {
$length = yield 2;
$length = \unpack("n", $length)[1];
$rawData = yield $length;
$callback($decoder->decode($rawData));
}
2017-06-23 07:34:11 +02:00
}
2019-01-25 02:27:47 +01:00
protected function __construct($socket)
{
2017-06-23 07:34:11 +02:00
parent::__construct($socket);
2017-06-23 07:34:11 +02:00
$this->encoder = (new EncoderFactory)->create();
$this->queue = new \SplQueue;
$this->parser = new Parser(self::parser([$this->queue, 'push']));
}
2019-01-25 02:27:47 +01:00
protected function send(Message $message): Promise
{
2017-06-23 07:34:11 +02:00
$data = $this->encoder->encode($message);
2017-06-23 12:49:16 +02:00
$promise = $this->write(\pack("n", \strlen($data)) . $data);
$promise->onResolve(function ($error) {
if ($error) {
$this->isAlive = false;
}
});
return $promise;
2017-06-23 07:34:11 +02:00
}
2019-01-25 02:27:47 +01:00
protected function receive(): Promise
{
2017-06-23 07:34:11 +02:00
if ($this->queue->isEmpty()) {
return call(function () {
do {
2017-06-29 12:36:09 +02:00
$chunk = yield $this->read();
2017-06-23 07:34:11 +02:00
if ($chunk === null) {
2017-06-23 12:49:16 +02:00
$this->isAlive = false;
throw new DnsException("Reading from the server failed");
2017-06-23 07:34:11 +02:00
}
$this->parser->push($chunk);
} while ($this->queue->isEmpty());
return $this->queue->shift();
});
}
return new Success($this->queue->shift());
}
2017-06-23 12:49:16 +02:00
2019-01-25 02:27:47 +01:00
public function isAlive(): bool
{
2017-06-23 12:49:16 +02:00
return $this->isAlive;
}
}