2016-12-30 06:21:17 +01:00
|
|
|
<?php
|
2016-09-14 16:27:39 +02:00
|
|
|
|
|
|
|
namespace Amp\Postgres;
|
|
|
|
|
2017-06-21 05:17:53 +02:00
|
|
|
use Amp\CallableMaker;
|
|
|
|
use Amp\Deferred;
|
|
|
|
use Amp\Emitter;
|
|
|
|
use Amp\Loop;
|
|
|
|
use Amp\Promise;
|
2017-07-28 06:20:16 +02:00
|
|
|
use Amp\Success;
|
2017-06-21 05:17:53 +02:00
|
|
|
use function Amp\call;
|
2016-09-14 16:27:39 +02:00
|
|
|
|
2017-08-01 07:38:12 +02:00
|
|
|
class PgSqlHandle implements Handle {
|
2016-09-14 16:27:39 +02:00
|
|
|
use CallableMaker;
|
2017-05-16 06:28:37 +02:00
|
|
|
|
2016-09-14 16:27:39 +02:00
|
|
|
/** @var resource PostgreSQL connection handle. */
|
|
|
|
private $handle;
|
|
|
|
|
|
|
|
/** @var \Amp\Deferred|null */
|
2016-09-21 07:18:24 +02:00
|
|
|
private $deferred;
|
2016-09-14 16:27:39 +02:00
|
|
|
|
|
|
|
/** @var string */
|
|
|
|
private $poll;
|
|
|
|
|
|
|
|
/** @var string */
|
|
|
|
private $await;
|
|
|
|
|
|
|
|
/** @var callable */
|
|
|
|
private $executeCallback;
|
2017-05-16 06:28:37 +02:00
|
|
|
|
2017-07-27 07:32:34 +02:00
|
|
|
/** @var callable */
|
|
|
|
private $deallocateCallback;
|
|
|
|
|
2017-01-18 18:05:05 +01:00
|
|
|
/** @var \Amp\Emitter[] */
|
2016-09-19 18:12:32 +02:00
|
|
|
private $listeners = [];
|
2017-05-16 06:28:37 +02:00
|
|
|
|
2016-09-19 18:12:32 +02:00
|
|
|
/** @var callable */
|
|
|
|
private $unlisten;
|
2017-05-16 06:28:37 +02:00
|
|
|
|
2017-07-28 06:20:16 +02:00
|
|
|
/** @var \Amp\Postgres\Internal\StatementStorage[] */
|
|
|
|
private $statements = [];
|
|
|
|
|
2016-09-14 16:27:39 +02:00
|
|
|
/**
|
|
|
|
* Connection constructor.
|
|
|
|
*
|
|
|
|
* @param resource $handle PostgreSQL connection handle.
|
|
|
|
* @param resource $socket PostgreSQL connection stream socket.
|
|
|
|
*/
|
|
|
|
public function __construct($handle, $socket) {
|
|
|
|
$this->handle = $handle;
|
|
|
|
|
2016-09-21 07:18:24 +02:00
|
|
|
$deferred = &$this->deferred;
|
2016-09-19 18:12:32 +02:00
|
|
|
$listeners = &$this->listeners;
|
2017-05-16 06:28:37 +02:00
|
|
|
|
2016-09-19 18:12:32 +02:00
|
|
|
$this->poll = Loop::onReadable($socket, static function ($watcher) use (&$deferred, &$listeners, $handle) {
|
2017-06-22 02:12:45 +02:00
|
|
|
if (!\pg_consume_input($handle)) {
|
|
|
|
if ($deferred !== null) {
|
|
|
|
$deferred->fail(new FailureException(\pg_last_error($handle)));
|
|
|
|
}
|
|
|
|
return;
|
|
|
|
}
|
2017-05-16 06:28:37 +02:00
|
|
|
|
2016-09-21 07:18:24 +02:00
|
|
|
while ($result = \pg_get_notify($handle, \PGSQL_ASSOC)) {
|
|
|
|
$channel = $result["message"];
|
2017-06-21 06:04:54 +02:00
|
|
|
|
|
|
|
if (!isset($listeners[$channel])) {
|
2017-08-02 06:26:27 +02:00
|
|
|
continue;
|
2016-09-19 18:12:32 +02:00
|
|
|
}
|
2017-06-21 06:04:54 +02:00
|
|
|
|
|
|
|
$notification = new Notification;
|
|
|
|
$notification->channel = $channel;
|
|
|
|
$notification->pid = $result["pid"];
|
|
|
|
$notification->payload = $result["payload"];
|
|
|
|
$listeners[$channel]->emit($notification);
|
2016-09-19 18:12:32 +02:00
|
|
|
}
|
2017-05-16 06:28:37 +02:00
|
|
|
|
2016-09-21 07:18:24 +02:00
|
|
|
if ($deferred === null) {
|
|
|
|
return; // No active query, only notification listeners.
|
|
|
|
}
|
2017-05-16 06:28:37 +02:00
|
|
|
|
2017-06-22 02:12:45 +02:00
|
|
|
if (\pg_connection_busy($handle)) {
|
|
|
|
return;
|
2016-09-21 07:18:24 +02:00
|
|
|
}
|
2016-09-14 16:27:39 +02:00
|
|
|
|
2017-06-22 02:12:45 +02:00
|
|
|
$deferred->resolve(\pg_get_result($handle));
|
|
|
|
|
2017-07-27 07:32:34 +02:00
|
|
|
if (!$deferred && empty($listeners)) {
|
2017-05-26 20:14:04 +02:00
|
|
|
Loop::disable($watcher);
|
2016-09-14 16:27:39 +02:00
|
|
|
}
|
|
|
|
});
|
|
|
|
|
|
|
|
$this->await = Loop::onWritable($socket, static function ($watcher) use (&$deferred, $handle) {
|
|
|
|
$flush = \pg_flush($handle);
|
2017-06-21 06:04:54 +02:00
|
|
|
if ($flush === 0) {
|
2016-09-14 16:27:39 +02:00
|
|
|
return; // Not finished sending data, listen again.
|
|
|
|
}
|
2017-05-16 06:28:37 +02:00
|
|
|
|
2016-09-14 16:27:39 +02:00
|
|
|
Loop::disable($watcher);
|
|
|
|
|
|
|
|
if ($flush === false) {
|
|
|
|
$deferred->fail(new FailureException(\pg_last_error($handle)));
|
|
|
|
}
|
|
|
|
});
|
2017-05-16 06:28:37 +02:00
|
|
|
|
2016-09-14 16:27:39 +02:00
|
|
|
Loop::disable($this->poll);
|
|
|
|
Loop::disable($this->await);
|
|
|
|
|
|
|
|
$this->executeCallback = $this->callableFromInstanceMethod("sendExecute");
|
2017-07-27 07:32:34 +02:00
|
|
|
$this->deallocateCallback = $this->callableFromInstanceMethod("sendDeallocate");
|
2016-09-19 18:12:32 +02:00
|
|
|
$this->unlisten = $this->callableFromInstanceMethod("unlisten");
|
2016-09-14 16:27:39 +02:00
|
|
|
}
|
|
|
|
|
|
|
|
/**
|
|
|
|
* Frees Io watchers from loop.
|
|
|
|
*/
|
|
|
|
public function __destruct() {
|
|
|
|
if (\is_resource($this->handle)) {
|
|
|
|
\pg_close($this->handle);
|
|
|
|
}
|
2017-05-16 06:28:37 +02:00
|
|
|
|
2016-09-14 16:27:39 +02:00
|
|
|
Loop::cancel($this->poll);
|
|
|
|
Loop::cancel($this->await);
|
|
|
|
}
|
|
|
|
|
|
|
|
/**
|
|
|
|
* @coroutine
|
|
|
|
*
|
|
|
|
* @param callable $function Function name to execute.
|
|
|
|
* @param mixed ...$args Arguments to pass to function.
|
|
|
|
*
|
|
|
|
* @return \Generator
|
|
|
|
*
|
|
|
|
* @resolve resource
|
|
|
|
*
|
|
|
|
* @throws \Amp\Postgres\FailureException
|
2017-05-26 20:14:04 +02:00
|
|
|
* @throws \Amp\Postgres\PendingOperationError
|
2016-09-14 16:27:39 +02:00
|
|
|
*/
|
|
|
|
private function send(callable $function, ...$args): \Generator {
|
2017-07-27 07:32:34 +02:00
|
|
|
while ($this->deferred) {
|
|
|
|
try {
|
|
|
|
yield $this->deferred->promise();
|
|
|
|
} catch (\Throwable $exception) {
|
|
|
|
// Ignore failure from another operation.
|
|
|
|
}
|
2016-09-14 16:27:39 +02:00
|
|
|
}
|
|
|
|
|
|
|
|
$result = $function($this->handle, ...$args);
|
|
|
|
|
|
|
|
if ($result === false) {
|
|
|
|
throw new FailureException(\pg_last_error($this->handle));
|
|
|
|
}
|
|
|
|
|
2016-09-21 07:18:24 +02:00
|
|
|
$this->deferred = new Deferred;
|
2016-09-14 16:27:39 +02:00
|
|
|
|
|
|
|
Loop::enable($this->poll);
|
|
|
|
if (0 === $result) {
|
|
|
|
Loop::enable($this->await);
|
|
|
|
}
|
|
|
|
|
|
|
|
try {
|
2016-11-15 18:06:21 +01:00
|
|
|
$result = yield $this->deferred->promise();
|
2016-09-14 16:27:39 +02:00
|
|
|
} finally {
|
2016-09-21 07:18:24 +02:00
|
|
|
$this->deferred = null;
|
2016-09-14 16:27:39 +02:00
|
|
|
}
|
|
|
|
|
|
|
|
return $result;
|
|
|
|
}
|
2017-05-16 06:28:37 +02:00
|
|
|
|
2016-09-14 16:27:39 +02:00
|
|
|
/**
|
|
|
|
* @param resource $result PostgreSQL result resource.
|
|
|
|
*
|
2017-05-16 06:14:02 +02:00
|
|
|
* @return \Amp\Postgres\CommandResult|\Amp\Postgres\TupleResult
|
2016-09-14 16:27:39 +02:00
|
|
|
*
|
|
|
|
* @throws \Amp\Postgres\FailureException
|
|
|
|
* @throws \Amp\Postgres\QueryError
|
|
|
|
*/
|
2017-05-16 06:14:02 +02:00
|
|
|
private function createResult($result) {
|
2016-09-14 16:27:39 +02:00
|
|
|
switch (\pg_result_status($result, \PGSQL_STATUS_LONG)) {
|
|
|
|
case \PGSQL_EMPTY_QUERY:
|
|
|
|
throw new QueryError("Empty query string");
|
|
|
|
|
|
|
|
case \PGSQL_COMMAND_OK:
|
|
|
|
return new PgSqlCommandResult($result);
|
|
|
|
|
|
|
|
case \PGSQL_TUPLES_OK:
|
|
|
|
return new PgSqlTupleResult($result);
|
|
|
|
|
|
|
|
case \PGSQL_NONFATAL_ERROR:
|
|
|
|
case \PGSQL_FATAL_ERROR:
|
|
|
|
throw new QueryError(\pg_result_error($result));
|
|
|
|
|
|
|
|
case \PGSQL_BAD_RESPONSE:
|
|
|
|
throw new FailureException(\pg_result_error($result));
|
|
|
|
|
|
|
|
default:
|
2017-07-31 07:34:05 +02:00
|
|
|
// @codeCoverageIgnoreStart
|
2016-09-14 16:27:39 +02:00
|
|
|
throw new FailureException("Unknown result status");
|
2017-07-31 07:34:05 +02:00
|
|
|
// @codeCoverageIgnoreEnd
|
2016-09-14 16:27:39 +02:00
|
|
|
}
|
|
|
|
}
|
2017-05-16 06:28:37 +02:00
|
|
|
|
2016-11-15 18:06:21 +01:00
|
|
|
private function sendExecute(string $name, array $params): Promise {
|
2017-05-16 06:14:02 +02:00
|
|
|
return call(function () use ($name, $params) {
|
|
|
|
return $this->createResult(yield from $this->send("pg_send_execute", $name, $params));
|
|
|
|
});
|
2016-09-14 16:27:39 +02:00
|
|
|
}
|
|
|
|
|
2017-07-28 06:20:16 +02:00
|
|
|
private function sendDeallocate(string $name) {
|
|
|
|
\assert(isset($this->statements[$name]), "Named statement not found when deallocating");
|
|
|
|
|
|
|
|
$storage = $this->statements[$name];
|
|
|
|
|
|
|
|
if (--$storage->count) {
|
|
|
|
return;
|
|
|
|
}
|
|
|
|
|
|
|
|
unset($this->statements[$name]);
|
|
|
|
|
|
|
|
Promise\rethrow($this->query(\sprintf("DEALLOCATE %s", $name)));
|
2017-07-27 07:32:34 +02:00
|
|
|
}
|
|
|
|
|
2016-09-14 16:27:39 +02:00
|
|
|
/**
|
|
|
|
* {@inheritdoc}
|
|
|
|
*/
|
2016-11-15 18:06:21 +01:00
|
|
|
public function query(string $sql): Promise {
|
2017-05-16 06:14:02 +02:00
|
|
|
return call(function () use ($sql) {
|
|
|
|
return $this->createResult(yield from $this->send("pg_send_query", $sql));
|
|
|
|
});
|
2016-09-14 16:27:39 +02:00
|
|
|
}
|
|
|
|
|
|
|
|
/**
|
|
|
|
* {@inheritdoc}
|
|
|
|
*/
|
2016-11-15 18:06:21 +01:00
|
|
|
public function execute(string $sql, ...$params): Promise {
|
2017-05-16 06:14:02 +02:00
|
|
|
return call(function () use ($sql, $params) {
|
2017-07-26 20:49:40 +02:00
|
|
|
return $this->createResult(yield from $this->send("pg_send_query_params", $sql, $params));
|
2017-05-16 06:14:02 +02:00
|
|
|
});
|
2016-09-14 16:27:39 +02:00
|
|
|
}
|
|
|
|
|
|
|
|
/**
|
|
|
|
* {@inheritdoc}
|
|
|
|
*/
|
2016-11-15 18:06:21 +01:00
|
|
|
public function prepare(string $sql): Promise {
|
2017-07-28 06:20:16 +02:00
|
|
|
$name = self::STATEMENT_NAME_PREFIX . \sha1($sql);
|
|
|
|
|
|
|
|
if (isset($this->statements[$name])) {
|
|
|
|
$storage = $this->statements[$name];
|
|
|
|
++$storage->count;
|
|
|
|
|
|
|
|
if ($storage->promise) {
|
|
|
|
return $storage->promise;
|
|
|
|
}
|
|
|
|
|
|
|
|
return new Success(new PgSqlStatement($name, $sql, $this->executeCallback, $this->deallocateCallback));
|
|
|
|
}
|
|
|
|
|
|
|
|
$this->statements[$name] = $storage = new Internal\StatementStorage;
|
|
|
|
|
2017-08-01 07:38:12 +02:00
|
|
|
$promise = $storage->promise = call(function () use ($name, $sql) {
|
2017-07-31 07:34:05 +02:00
|
|
|
/** @var resource $result PostgreSQL result resource. */
|
|
|
|
$result = yield from $this->send("pg_send_prepare", $name, $sql);
|
|
|
|
|
|
|
|
switch (\pg_result_status($result, \PGSQL_STATUS_LONG)) {
|
|
|
|
case \PGSQL_COMMAND_OK:
|
|
|
|
return new PgSqlStatement($name, $sql, $this->executeCallback, $this->deallocateCallback);
|
|
|
|
|
|
|
|
case \PGSQL_NONFATAL_ERROR:
|
|
|
|
case \PGSQL_FATAL_ERROR:
|
|
|
|
throw new QueryError(\pg_result_error($result));
|
|
|
|
|
|
|
|
case \PGSQL_BAD_RESPONSE:
|
|
|
|
throw new FailureException(\pg_result_error($result));
|
|
|
|
|
|
|
|
default:
|
|
|
|
// @codeCoverageIgnoreStart
|
|
|
|
throw new FailureException("Unknown result status");
|
|
|
|
// @codeCoverageIgnoreEnd
|
|
|
|
}
|
2016-09-14 16:27:39 +02:00
|
|
|
});
|
2017-08-01 07:38:12 +02:00
|
|
|
$promise->onResolve(function ($exception) use ($storage, $name) {
|
2017-07-31 07:52:56 +02:00
|
|
|
if ($exception) {
|
|
|
|
unset($this->statements[$name]);
|
|
|
|
return;
|
|
|
|
}
|
|
|
|
|
2017-07-28 06:20:16 +02:00
|
|
|
$storage->promise = null;
|
|
|
|
});
|
2017-08-01 07:38:12 +02:00
|
|
|
return $promise;
|
2016-09-14 16:27:39 +02:00
|
|
|
}
|
2017-05-16 06:28:37 +02:00
|
|
|
|
2016-09-21 07:18:24 +02:00
|
|
|
/**
|
|
|
|
* {@inheritdoc}
|
|
|
|
*/
|
2016-11-15 18:06:21 +01:00
|
|
|
public function notify(string $channel, string $payload = ""): Promise {
|
2016-09-21 07:18:24 +02:00
|
|
|
if ($payload === "") {
|
2017-08-01 07:38:12 +02:00
|
|
|
return $this->query(\sprintf("NOTIFY %s", $this->quoteName($channel)));
|
2016-09-21 07:18:24 +02:00
|
|
|
}
|
2017-05-16 06:28:37 +02:00
|
|
|
|
2017-08-01 07:38:12 +02:00
|
|
|
return $this->query(\sprintf("NOTIFY %s, %s", $this->quoteName($channel), $this->quoteString($payload)));
|
2016-09-21 07:18:24 +02:00
|
|
|
}
|
2017-05-16 06:28:37 +02:00
|
|
|
|
2016-09-19 18:12:32 +02:00
|
|
|
/**
|
|
|
|
* {@inheritdoc}
|
|
|
|
*/
|
2016-11-15 18:06:21 +01:00
|
|
|
public function listen(string $channel): Promise {
|
2017-05-16 06:14:02 +02:00
|
|
|
return call(function () use ($channel) {
|
2017-06-13 07:34:20 +02:00
|
|
|
if (isset($this->listeners[$channel])) {
|
|
|
|
throw new QueryError(\sprintf("Already listening on channel '%s'", $channel));
|
|
|
|
}
|
|
|
|
|
2017-06-21 06:04:54 +02:00
|
|
|
$this->listeners[$channel] = $emitter = new Emitter;
|
2017-06-13 07:34:20 +02:00
|
|
|
|
|
|
|
try {
|
2017-08-01 07:38:12 +02:00
|
|
|
yield $this->query(\sprintf("LISTEN %s", $this->quoteName($channel)));
|
2017-06-13 07:34:20 +02:00
|
|
|
} catch (\Throwable $exception) {
|
|
|
|
unset($this->listeners[$channel]);
|
|
|
|
throw $exception;
|
|
|
|
}
|
2017-05-16 06:14:02 +02:00
|
|
|
|
2016-09-21 07:18:24 +02:00
|
|
|
Loop::enable($this->poll);
|
2017-05-16 06:14:02 +02:00
|
|
|
return new Listener($emitter->iterate(), $channel, $this->unlisten);
|
2016-09-19 18:12:32 +02:00
|
|
|
});
|
|
|
|
}
|
2017-05-16 06:28:37 +02:00
|
|
|
|
2016-09-19 18:12:32 +02:00
|
|
|
/**
|
|
|
|
* @param string $channel
|
|
|
|
*
|
2017-03-17 16:17:24 +01:00
|
|
|
* @return \Amp\Promise
|
2016-09-19 18:12:32 +02:00
|
|
|
*
|
|
|
|
* @throws \Error
|
|
|
|
*/
|
2016-11-15 18:06:21 +01:00
|
|
|
private function unlisten(string $channel): Promise {
|
2017-07-28 06:20:16 +02:00
|
|
|
\assert(isset($this->listeners[$channel]), "Not listening on that channel");
|
2017-05-16 06:28:37 +02:00
|
|
|
|
2017-01-18 19:29:48 +01:00
|
|
|
$emitter = $this->listeners[$channel];
|
2016-09-19 18:12:32 +02:00
|
|
|
unset($this->listeners[$channel]);
|
2017-05-16 06:28:37 +02:00
|
|
|
|
2016-09-21 07:18:24 +02:00
|
|
|
if (empty($this->listeners) && $this->deferred === null) {
|
|
|
|
Loop::disable($this->poll);
|
|
|
|
}
|
2017-05-16 06:28:37 +02:00
|
|
|
|
2017-08-01 07:38:12 +02:00
|
|
|
$promise = $this->query(\sprintf("UNLISTEN %s", $this->quoteName($channel)));
|
2017-06-21 06:04:54 +02:00
|
|
|
$promise->onResolve([$emitter, "complete"]);
|
2016-11-15 18:06:21 +01:00
|
|
|
return $promise;
|
2016-09-19 18:12:32 +02:00
|
|
|
}
|
2017-08-01 07:38:12 +02:00
|
|
|
|
|
|
|
/**
|
|
|
|
* {@inheritdoc}
|
|
|
|
*/
|
|
|
|
public function quoteString(string $data): string {
|
|
|
|
return \pg_escape_literal($this->handle, $data);
|
|
|
|
}
|
|
|
|
|
|
|
|
/**
|
|
|
|
* {@inheritdoc}
|
|
|
|
*/
|
|
|
|
public function quoteName(string $name): string {
|
|
|
|
return \pg_escape_identifier($this->handle, $name);
|
|
|
|
}
|
2016-09-14 16:27:39 +02:00
|
|
|
}
|