danog/postgres
1
0
Fork 0
mirror of https://github.com/danog/postgres.git synced 2025-01-22 05:11:14 +01:00
Code Issues Projects Releases Wiki Activity

Awaitable → Promise

Browse Source
This commit is contained in:
Aaron Piotrowski 2016-11-15 11:06:21 -06:00
parent 492d059a52
commit 0fb3f9bfdb
19 changed files with 141 additions and 141 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

30
lib/AbstractConnection.php
View File

@ -3,7 +3,7 @@
namespace Amp\Postgres;
use Amp\{ CallableMaker, Coroutine, Deferred, function pipe };
use Interop\Async\Awaitable;
use Interop\Async\Promise;
abstract class AbstractConnection implements Connection {
use CallableMaker;
@ -21,9 +21,9 @@ abstract class AbstractConnection implements Connection {
* @param string $connectionString
* @param int $timeout Timeout until the connection attempt fails.
*
* @return \Interop\Async\Awaitable<\Amp\Postgres\Connection>
* @return \Interop\Async\Promise<\Amp\Postgres\Connection>
*/
abstract public static function connect(string $connectionString, int $timeout = null): Awaitable;
abstract public static function connect(string $connectionString, int $timeout = null): Promise;
/**
* @param $executor;
@ -43,7 +43,7 @@ abstract class AbstractConnection implements Connection {
*/
private function send(callable $method, ...$args): \Generator {
while ($this->busy !== null) {
yield $this->busy->getAwaitable();
yield $this->busy->promise();
}
return $method(...$args);
@ -61,21 +61,21 @@ abstract class AbstractConnection implements Connection {
/**
* {@inheritdoc}
*/
public function query(string $sql): Awaitable {
public function query(string $sql): Promise {
return new Coroutine($this->send([$this->executor, "query"], $sql));
}
/**
* {@inheritdoc}
*/
public function execute(string $sql, ...$params): Awaitable {
public function execute(string $sql, ...$params): Promise {
return new Coroutine($this->send([$this->executor, "execute"], $sql, ...$params));
}
/**
* {@inheritdoc}
*/
public function prepare(string $sql): Awaitable {
public function prepare(string $sql): Promise {
return new Coroutine($this->send([$this->executor, "prepare"], $sql));
}
@ -83,43 +83,43 @@ abstract class AbstractConnection implements Connection {
/**
* {@inheritdoc}
*/
public function notify(string $channel, string $payload = ""): Awaitable {
public function notify(string $channel, string $payload = ""): Promise {
return new Coroutine($this->send([$this->executor, "notify"], $channel, $payload));
}
/**
* {@inheritdoc}
*/
public function listen(string $channel): Awaitable {
public function listen(string $channel): Promise {
return new Coroutine($this->send([$this->executor, "listen"], $channel));
}
/**
* {@inheritdoc}
*/
public function transaction(int $isolation = Transaction::COMMITTED): Awaitable {
public function transaction(int $isolation = Transaction::COMMITTED): Promise {
switch ($isolation) {
case Transaction::UNCOMMITTED:
$awaitable = $this->query("BEGIN TRANSACTION ISOLATION LEVEL READ UNCOMMITTED");
$promise = $this->query("BEGIN TRANSACTION ISOLATION LEVEL READ UNCOMMITTED");
break;
case Transaction::COMMITTED:
$awaitable = $this->query("BEGIN TRANSACTION ISOLATION LEVEL READ COMMITTED");
$promise = $this->query("BEGIN TRANSACTION ISOLATION LEVEL READ COMMITTED");
break;
case Transaction::REPEATABLE:
$awaitable = $this->query("BEGIN TRANSACTION ISOLATION LEVEL REPEATABLE READ");
$promise = $this->query("BEGIN TRANSACTION ISOLATION LEVEL REPEATABLE READ");
break;
case Transaction::SERIALIZABLE:
$awaitable = $this->query("BEGIN TRANSACTION ISOLATION LEVEL SERIALIZABLE");
$promise = $this->query("BEGIN TRANSACTION ISOLATION LEVEL SERIALIZABLE");
break;
default:
throw new \Error("Invalid transaction type");
}
return pipe($awaitable, function (CommandResult $result) use ($isolation) {
return pipe($promise, function (CommandResult $result) use ($isolation) {
$this->busy = new Deferred;
$transaction = new Transaction($this->executor, $isolation);
$transaction->onComplete($this->release);

38
lib/AbstractPool.php
View File

@ -3,7 +3,7 @@
namespace Amp\Postgres;
use Amp\{ Coroutine, Deferred };
use Interop\Async\Awaitable;
use Interop\Async\Promise;
abstract class AbstractPool implements Pool {
/** @var \SplQueue */
@ -15,24 +15,24 @@ abstract class AbstractPool implements Pool {
/** @var \SplObjectStorage */
private $connections;
/** @var \Interop\Async\Awaitable|null */
private $awaitable;
/** @var \Interop\Async\Promise|null */
private $promise;
/** @var \Amp\Deferred|null */
private $deferred;
/** @var \Amp\Postgres\Connection|\Interop\Async\Awaitable|null Connection used for notification listening. */
/** @var \Amp\Postgres\Connection|\Interop\Async\Promise|null Connection used for notification listening. */
private $listeningConnection;
/** @var int Number of listeners on listening connection. */
private $listenerCount = 0;
/**
* @return \Interop\Async\Awaitable<\Amp\Postgres\Connection>
* @return \Interop\Async\Promise<\Amp\Postgres\Connection>
*
* @throws \Amp\Postgres\FailureException
*/
abstract protected function createConnection(): Awaitable;
abstract protected function createConnection(): Promise;
public function __construct() {
$this->connections = new \SplObjectStorage();
@ -74,9 +74,9 @@ abstract class AbstractPool implements Pool {
* @resolve \Amp\Postgres\Connection
*/
private function pop(): \Generator {
while ($this->awaitable !== null) {
while ($this->promise !== null) {
try {
yield $this->awaitable; // Prevent simultaneous connection creation.
yield $this->promise; // Prevent simultaneous connection creation.
} catch (\Throwable $exception) {
// Ignore failure or cancellation of other operations.
}
@ -87,15 +87,15 @@ abstract class AbstractPool implements Pool {
if ($this->connections->count() >= $this->getMaxConnections()) {
// All possible connections busy, so wait until one becomes available.
$this->deferred = new Deferred;
yield $this->awaitable = $this->deferred->getAwaitable();
yield $this->promise = $this->deferred->promise();
} else {
// Max connection count has not been reached, so open another connection.
$this->awaitable = $this->createConnection();
$this->addConnection(yield $this->awaitable);
$this->promise = $this->createConnection();
$this->addConnection(yield $this->promise);
}
} finally {
$this->deferred = null;
$this->awaitable = null;
$this->promise = null;
}
}
@ -123,7 +123,7 @@ abstract class AbstractPool implements Pool {
/**
* {@inheritdoc}
*/
public function query(string $sql): Awaitable {
public function query(string $sql): Promise {
return new Coroutine($this->doQuery($sql));
}
@ -152,7 +152,7 @@ abstract class AbstractPool implements Pool {
/**
* {@inheritdoc}
*/
public function execute(string $sql, ...$params): Awaitable {
public function execute(string $sql, ...$params): Promise {
return new Coroutine($this->doExecute($sql, $params));
}
@ -181,7 +181,7 @@ abstract class AbstractPool implements Pool {
/**
* {@inheritdoc}
*/
public function prepare(string $sql): Awaitable {
public function prepare(string $sql): Promise {
return new Coroutine($this->doPrepare($sql));
}
@ -202,7 +202,7 @@ abstract class AbstractPool implements Pool {
/**
* {@inheritdoc}
*/
public function notify(string $channel, string $payload = ""): Awaitable {
public function notify(string $channel, string $payload = ""): Promise {
return new Coroutine($this->doNotify($channel, $payload));
}
@ -222,7 +222,7 @@ abstract class AbstractPool implements Pool {
/**
* {@inheritdoc}
*/
public function listen(string $channel): Awaitable {
public function listen(string $channel): Promise {
return new Coroutine($this->doListen($channel));
}
@ -233,7 +233,7 @@ abstract class AbstractPool implements Pool {
$this->listeningConnection = new Coroutine($this->pop());
}
if ($this->listeningConnection instanceof Awaitable) {
if ($this->listeningConnection instanceof Promise) {
$this->listeningConnection = yield $this->listeningConnection;
}
@ -263,7 +263,7 @@ abstract class AbstractPool implements Pool {
/**
* {@inheritdoc}
*/
public function transaction(int $isolation = Transaction::COMMITTED): Awaitable {
public function transaction(int $isolation = Transaction::COMMITTED): Promise {
return new Coroutine($this->doTransaction($isolation));
}

4
lib/AggregatePool.php
View File

@ -2,7 +2,7 @@
namespace Amp\Postgres;
use Interop\Async\Awaitable;
use Interop\Async\Promise;
class AggregatePool extends AbstractPool {
/**
@ -15,7 +15,7 @@ class AggregatePool extends AbstractPool {
/**
* {@inheritdoc}
*/
protected function createConnection(): Awaitable {
protected function createConnection(): Promise {
throw new PoolError("Creating connections is not available in an aggregate pool");
}

10
lib/Connection.php
View File

@ -2,24 +2,24 @@
namespace Amp\Postgres;
use Interop\Async\Awaitable;
use Interop\Async\Promise;
interface Connection extends Executor {
/**
* @param int $isolation
*
* @return \Interop\Async\Awaitable<\Amp\Postgres\Transaction>
* @return \Interop\Async\Promise<\Amp\Postgres\Transaction>
*
* @throws \Amp\Postgres\FailureException
*/
public function transaction(int $isolation = Transaction::COMMITTED): Awaitable;
public function transaction(int $isolation = Transaction::COMMITTED): Promise;
/**
* @param string $channel Channel name.
*
* @return \Interop\Async\Awaitable<\Amp\Postgres\Listener>
* @return \Interop\Async\Promise<\Amp\Postgres\Listener>
*
* @throws \Amp\Postgres\FailureException
*/
public function listen(string $channel): Awaitable;
public function listen(string $channel): Promise;
}

4
lib/ConnectionPool.php
View File

@ -2,7 +2,7 @@
namespace Amp\Postgres;
use Interop\Async\Awaitable;
use Interop\Async\Promise;
class ConnectionPool extends AbstractPool {
const DEFAULT_MAX_CONNECTIONS = 100;
@ -41,7 +41,7 @@ class ConnectionPool extends AbstractPool {
/**
* {@inheritdoc}
*/
protected function createConnection(): Awaitable {
protected function createConnection(): Promise {
return connect($this->connectionString, $this->connectTimeout);
}

18
lib/Executor.php
View File

@ -2,42 +2,42 @@
namespace Amp\Postgres;
use Interop\Async\Awaitable;
use Interop\Async\Promise;
interface Executor {
/**
* @param string $sql
*
* @return \Interop\Async\Awaitable<\Amp\Postgres\Result>
* @return \Interop\Async\Promise<\Amp\Postgres\Result>
*
* @throws \Amp\Postgres\FailureException
*/
public function query(string $sql): Awaitable;
public function query(string $sql): Promise;
/**
* @param string $sql
* @param mixed ...$params
*
* @return \Interop\Async\Awaitable<\Amp\Postgres\Result>
* @return \Interop\Async\Promise<\Amp\Postgres\Result>
*
* @throws \Amp\Postgres\FailureException
*/
public function execute(string $sql, ...$params): Awaitable;
public function execute(string $sql, ...$params): Promise;
/**
* @param string $sql
*
* @return \Interop\Async\Awaitable<\Amp\Postgres\Statement>
* @return \Interop\Async\Promise<\Amp\Postgres\Statement>
*
* @throws \Amp\Postgres\FailureException
*/
public function prepare(string $sql): Awaitable;
public function prepare(string $sql): Promise;
/**
* @param string $channel Channel name.
* @param string $payload Notification payload.
*
* @return \Interop\Async\Awaitable<\Amp\Postgres\CommandResult>
* @return \Interop\Async\Promise<\Amp\Postgres\CommandResult>
*/
public function notify(string $channel, string $payload = ""): Awaitable;
public function notify(string $channel, string $payload = ""): Promise;
}

12
lib/Listener.php
View File

@ -3,7 +3,7 @@
namespace Amp\Postgres;
use Amp\{ Observable, Observer };
use Interop\Async\Awaitable;
use Interop\Async\Promise;
class Listener extends Observer implements Operation {
use Internal\Operation;
@ -35,13 +35,13 @@ class Listener extends Observer implements Operation {
/**
* Unlistens from the channel. No more values will be emitted on theis channel.
*
* @return \Interop\Async\Awaitable<\Amp\Postgres\CommandResult>
* @return \Interop\Async\Promise<\Amp\Postgres\CommandResult>
*/
public function unlisten(): Awaitable {
$awaitable = ($this->unlisten)($this->channel);
$awaitable->when(function () {
public function unlisten(): Promise {
$promise = ($this->unlisten)($this->channel);
$promise->when(function () {
$this->complete();
});
return $awaitable;
return $promise;
}
}

10
lib/PgSqlConnection.php
View File

@ -3,18 +3,18 @@
namespace Amp\Postgres;
use Amp\{ Deferred, TimeoutException };
use Interop\Async\{ Awaitable, Loop };
use Interop\Async\{ Loop, Promise };
class PgSqlConnection extends AbstractConnection {
/**
* @param string $connectionString
* @param int|null $timeout
*
* @return \Interop\Async\Awaitable<\Amp\Postgres\PgSqlConnection>
* @return \Interop\Async\Promise<\Amp\Postgres\PgSqlConnection>
*
* @throws \Amp\Postgres\FailureException
*/
public static function connect(string $connectionString, int $timeout = null): Awaitable {
public static function connect(string $connectionString, int $timeout = null): Promise {
if (!$connection = @\pg_connect($connectionString, \PGSQL_CONNECT_ASYNC | \PGSQL_CONNECT_FORCE_NEW)) {
throw new FailureException("Failed to create connection resource");
}
@ -60,7 +60,7 @@ class PgSqlConnection extends AbstractConnection {
if ($timeout !== null) {
return \Amp\capture(
$deferred->getAwaitable(),
$deferred->promise(),
TimeoutException::class,
function (\Throwable $exception) use ($connection, $poll, $await) {
Loop::cancel($poll);
@ -71,7 +71,7 @@ class PgSqlConnection extends AbstractConnection {
);
}
return $deferred->getAwaitable();
return $deferred->promise();
}
/**

28
lib/PgSqlExecutor.php
View File

@ -3,7 +3,7 @@
namespace Amp\Postgres;
use Amp\{ CallableMaker, Coroutine, Deferred, Postponed, function pipe };
use Interop\Async\{ Awaitable, Loop };
use Interop\Async\{ Loop, Promise };
class PgSqlExecutor implements Executor {
use CallableMaker;
@ -124,7 +124,7 @@ class PgSqlExecutor implements Executor {
private function send(callable $function, ...$args): \Generator {
while ($this->deferred !== null) {
try {
yield $this->deferred->getAwaitable();
yield $this->deferred->promise();
} catch (\Throwable $exception) {
// Ignore failure from another operation.
}
@ -144,7 +144,7 @@ class PgSqlExecutor implements Executor {
}
try {
$result = yield $this->deferred->getAwaitable();
$result = yield $this->deferred->promise();
} finally {
$this->deferred = null;
}
@ -183,28 +183,28 @@ class PgSqlExecutor implements Executor {
}
}
private function sendExecute(string $name, array $params): Awaitable {
private function sendExecute(string $name, array $params): Promise {
return pipe(new Coroutine($this->send("pg_send_execute", $name, $params)), $this->createResult);
}
/**
* {@inheritdoc}
*/
public function query(string $sql): Awaitable {
public function query(string $sql): Promise {
return pipe(new Coroutine($this->send("pg_send_query", $sql)), $this->createResult);
}
/**
* {@inheritdoc}
*/
public function execute(string $sql, ...$params): Awaitable {
public function execute(string $sql, ...$params): Promise {
return pipe(new Coroutine($this->send("pg_send_query_params", $sql, $params)), $this->createResult);
}
/**
* {@inheritdoc}
*/
public function prepare(string $sql): Awaitable {
public function prepare(string $sql): Promise {
return pipe(new Coroutine($this->send("pg_send_prepare", $sql, $sql)), function () use ($sql) {
return new PgSqlStatement($sql, $this->executeCallback);
});
@ -213,7 +213,7 @@ class PgSqlExecutor implements Executor {
/**
* {@inheritdoc}
*/
public function notify(string $channel, string $payload = ""): Awaitable {
public function notify(string $channel, string $payload = ""): Promise {
if ($payload === "") {
return $this->query(\sprintf("NOTIFY %s"));
}
@ -224,7 +224,7 @@ class PgSqlExecutor implements Executor {
/**
* {@inheritdoc}
*/
public function listen(string $channel): Awaitable {
public function listen(string $channel): Promise {
return pipe($this->query(\sprintf("LISTEN %s", $channel)), function (CommandResult $result) use ($channel) {
$postponed = new Postponed;
$this->listeners[$channel] = $postponed;
@ -236,11 +236,11 @@ class PgSqlExecutor implements Executor {
/**
* @param string $channel
*
* @return \Interop\Async\Awaitable
* @return \Interop\Async\Promise
*
* @throws \Error
*/
private function unlisten(string $channel): Awaitable {
private function unlisten(string $channel): Promise {
if (!isset($this->listeners[$channel])) {
throw new \Error("Not listening on that channel");
}
@ -252,10 +252,10 @@ class PgSqlExecutor implements Executor {
Loop::disable($this->poll);
}
$awaitable = $this->query(\sprintf("UNLISTEN %s", $channel));
$awaitable->when(function () use ($postponed) {
$promise = $this->query(\sprintf("UNLISTEN %s", $channel));
$promise->when(function () use ($postponed) {
$postponed->resolve();
});
return $awaitable;
return $promise;
}
}

6
lib/PgSqlStatement.php
View File

@ -2,7 +2,7 @@
namespace Amp\Postgres;
use Interop\Async\Awaitable;
use Interop\Async\Promise;
class PgSqlStatement implements Statement {
/** @var string */
@ -30,11 +30,11 @@ class PgSqlStatement implements Statement {
/**
* @param mixed ...$params
*
* @return \Interop\Async\Awaitable<\Amp\Postgres\Result>
* @return \Interop\Async\Promise<\Amp\Postgres\Result>
*
* @throws \Amp\Postgres\FailureException If executing the statement fails.
*/
public function execute(...$params): Awaitable {
public function execute(...$params): Promise {
return ($this->execute)($this->sql, $params);
}
}

10
lib/PqConnection.php
View File

@ -3,7 +3,7 @@
namespace Amp\Postgres;
use Amp\{ Deferred, TimeoutException };
use Interop\Async\{ Awaitable, Loop };
use Interop\Async\{ Loop, Promise };
use pq;
class PqConnection extends AbstractConnection {
@ -11,11 +11,11 @@ class PqConnection extends AbstractConnection {
* @param string $connectionString
* @param int|null $timeout
*
* @return \Interop\Async\Awaitable<\Amp\Postgres\PgSqlConnection>
* @return \Interop\Async\Promise<\Amp\Postgres\PgSqlConnection>
*
* @throws \Amp\Postgres\FailureException
*/
public static function connect(string $connectionString, int $timeout = null): Awaitable {
public static function connect(string $connectionString, int $timeout = null): Promise {
try {
$connection = new pq\Connection($connectionString, pq\Connection::ASYNC);
} catch (pq\Exception $exception) {
@ -58,7 +58,7 @@ class PqConnection extends AbstractConnection {
if ($timeout !== null) {
return \Amp\capture(
$deferred->getAwaitable(),
$deferred->promise(),
TimeoutException::class,
function (\Throwable $exception) use ($connection, $poll, $await) {
Loop::cancel($poll);
@ -68,7 +68,7 @@ class PqConnection extends AbstractConnection {
);
}
return $deferred->getAwaitable();
return $deferred->promise();
}
/**

32
lib/PqExecutor.php
View File

@ -3,7 +3,7 @@
namespace Amp\Postgres;
use Amp\{ CallableMaker, Coroutine, Deferred, Postponed, function pipe };
use Interop\Async\{ Awaitable, Loop };
use Interop\Async\{ Loop, Promise };
use pq;
class PqExecutor implements Executor {
@ -107,7 +107,7 @@ class PqExecutor implements Executor {
*/
private function send(callable $method, ...$args): \Generator {
while ($this->busy !== null) {
yield $this->busy->getAwaitable();
yield $this->busy->promise();
}
$this->busy = new Deferred;
@ -127,7 +127,7 @@ class PqExecutor implements Executor {
}
try {
$result = yield $this->deferred->getAwaitable();
$result = yield $this->deferred->promise();
} finally {
$this->deferred = null;
}
@ -180,7 +180,7 @@ class PqExecutor implements Executor {
Loop::enable($this->poll);
try {
$result = yield $this->deferred->getAwaitable();
$result = yield $this->deferred->promise();
} finally {
$this->deferred = null;
}
@ -207,37 +207,37 @@ class PqExecutor implements Executor {
/**
* {@inheritdoc}
*/
public function query(string $sql): Awaitable {
public function query(string $sql): Promise {
return new Coroutine($this->send([$this->handle, "execAsync"], $sql));
}
/**
* {@inheritdoc}
*/
public function execute(string $sql, ...$params): Awaitable {
public function execute(string $sql, ...$params): Promise {
return new Coroutine($this->send([$this->handle, "execParamsAsync"], $sql, $params));
}
/**
* {@inheritdoc}
*/
public function prepare(string $sql): Awaitable {
public function prepare(string $sql): Promise {
return new Coroutine($this->send([$this->handle, "prepareAsync"], $sql, $sql));
}
/**
* {@inheritdoc}
*/
public function notify(string $channel, string $payload = ""): Awaitable {
public function notify(string $channel, string $payload = ""): Promise {
return new Coroutine($this->send([$this->handle, "notifyAsync"], $channel, $payload));
}
/**
* {@inheritdoc}
*/
public function listen(string $channel): Awaitable {
public function listen(string $channel): Promise {
$postponed = new Postponed;
$awaitable = new Coroutine($this->send(
$promise = new Coroutine($this->send(
[$this->handle, "listenAsync"],
$channel,
static function (string $channel, string $message, int $pid) use ($postponed) {
@ -248,7 +248,7 @@ class PqExecutor implements Executor {
$postponed->emit($notification);
}));
return pipe($awaitable, function () use ($postponed, $channel) {
return pipe($promise, function () use ($postponed, $channel) {
$this->listeners[$channel] = $postponed;
Loop::enable($this->poll);
return new Listener($postponed->getObservable(), $channel, $this->unlisten);
@ -258,11 +258,11 @@ class PqExecutor implements Executor {
/**
* @param string $channel
*
* @return \Interop\Async\Awaitable
* @return \Interop\Async\Promise
*
* @throws \Error
*/
private function unlisten(string $channel): Awaitable {
private function unlisten(string $channel): Promise {
if (!isset($this->listeners[$channel])) {
throw new \Error("Not listening on that channel");
}
@ -274,10 +274,10 @@ class PqExecutor implements Executor {
Loop::disable($this->poll);
}
$awaitable = new Coroutine($this->send([$this->handle, "unlistenAsync"], $channel));
$awaitable->when(function () use ($postponed) {
$promise = new Coroutine($this->send([$this->handle, "unlistenAsync"], $channel));
$promise->when(function () use ($postponed) {
$postponed->resolve();
});
return $awaitable;
return $promise;
}
}

6
lib/PqStatement.php
View File

@ -3,7 +3,7 @@
namespace Amp\Postgres;
use Amp\{ Coroutine, function rethrow };
use Interop\Async\Awaitable;
use Interop\Async\Promise;
use pq;
class PqStatement implements Statement {
@ -36,11 +36,11 @@ class PqStatement implements Statement {
/**
* @param mixed ...$params
*
* @return \Interop\Async\Awaitable<\Amp\Postgres\Result>
* @return \Interop\Async\Promise<\Amp\Postgres\Result>
*
* @throws \Amp\Postgres\FailureException If executing the statement fails.
*/
public function execute(...$params): Awaitable {
public function execute(...$params): Promise {
return new Coroutine(($this->execute)([$this->statement, "execAsync"], $params));
}
}

6
lib/Statement.php
View File

@ -2,13 +2,13 @@
namespace Amp\Postgres;
use Interop\Async\Awaitable;
use Interop\Async\Promise;
interface Statement {
/**
* @param mixed ...$params
*
* @return \Interop\Async\Awaitable<\Amp\Postgres\Result>
* @return \Interop\Async\Promise<\Amp\Postgres\Result>
*/
public function execute(...$params): Awaitable;
public function execute(...$params): Promise;
}

30
lib/Transaction.php
View File

@ -3,7 +3,7 @@
namespace Amp\Postgres;
use Amp\Coroutine;
use Interop\Async\Awaitable;
use Interop\Async\Promise;
class Transaction implements Executor, Operation {
use Internal\Operation;
@ -58,7 +58,7 @@ class Transaction implements Executor, Operation {
/**
* {@inheritdoc}
*/
public function query(string $sql): Awaitable {
public function query(string $sql): Promise {
if ($this->executor === null) {
throw new TransactionError("The transaction has been committed or rolled back");
}
@ -69,7 +69,7 @@ class Transaction implements Executor, Operation {
/**
* {@inheritdoc}
*/
public function prepare(string $sql): Awaitable {
public function prepare(string $sql): Promise {
if ($this->executor === null) {
throw new TransactionError("The transaction has been committed or rolled back");
}
@ -80,7 +80,7 @@ class Transaction implements Executor, Operation {
/**
* {@inheritdoc}
*/
public function execute(string $sql, ...$params): Awaitable {
public function execute(string $sql, ...$params): Promise {
if ($this->executor === null) {
throw new TransactionError("The transaction has been committed or rolled back");
}
@ -92,7 +92,7 @@ class Transaction implements Executor, Operation {
/**
* {@inheritdoc}
*/
public function notify(string $channel, string $payload = ""): Awaitable {
public function notify(string $channel, string $payload = ""): Promise {
if ($this->executor === null) {
throw new TransactionError("The transaction has been committed or rolled back");
}
@ -103,11 +103,11 @@ class Transaction implements Executor, Operation {
/**
* Commits the transaction and makes it inactive.
*
* @return \Interop\Async\Awaitable<\Amp\Postgres\CommandResult>
* @return \Interop\Async\Promise<\Amp\Postgres\CommandResult>
*
* @throws \Amp\Postgres\TransactionError
*/
public function commit(): Awaitable {
public function commit(): Promise {
return new Coroutine($this->doCommit());
}
@ -131,11 +131,11 @@ class Transaction implements Executor, Operation {
/**
* Rolls back the transaction and makes it inactive.
*
* @return \Interop\Async\Awaitable<\Amp\Postgres\CommandResult>
* @return \Interop\Async\Promise<\Amp\Postgres\CommandResult>
*
* @throws \Amp\Postgres\TransactionError
*/
public function rollback(): Awaitable {
public function rollback(): Promise {
return new Coroutine($this->doRollback());
}
@ -161,11 +161,11 @@ class Transaction implements Executor, Operation {
*
* @param string $identifier Savepoint identifier.
*
* @return \Interop\Async\Awaitable<\Amp\Postgres\CommandResult>
* @return \Interop\Async\Promise<\Amp\Postgres\CommandResult>
*
* @throws \Amp\Postgres\TransactionError
*/
public function savepoint(string $identifier): Awaitable {
public function savepoint(string $identifier): Promise {
return $this->query("SAVEPOINT " . $identifier);
}
@ -175,11 +175,11 @@ class Transaction implements Executor, Operation {
*
* @param string $identifier Savepoint identifier.
*
* @return \Interop\Async\Awaitable<\Amp\Postgres\CommandResult>
* @return \Interop\Async\Promise<\Amp\Postgres\CommandResult>
*
* @throws \Amp\Postgres\TransactionError
*/
public function rollbackTo(string $identifier): Awaitable {
public function rollbackTo(string $identifier): Promise {
return $this->query("ROLLBACK TO " . $identifier);
}
@ -189,11 +189,11 @@ class Transaction implements Executor, Operation {
*
* @param string $identifier Savepoint identifier.
*
* @return \Interop\Async\Awaitable<\Amp\Postgres\CommandResult>
* @return \Interop\Async\Promise<\Amp\Postgres\CommandResult>
*
* @throws \Amp\Postgres\TransactionError
*/
public function release(string $identifier): Awaitable {
public function release(string $identifier): Promise {
return $this->query("RELEASE SAVEPOINT " . $identifier);
}
}

6
lib/functions.php
View File

@ -2,18 +2,18 @@
namespace Amp\Postgres;
use Interop\Async\Awaitable;
use Interop\Async\Promise;
/**
* @param string $connectionString
* @param int $timeout
*
* @return \Interop\Async\Awaitable<\Amp\Postgres\Connection>
* @return \Interop\Async\Promise<\Amp\Postgres\Connection>
*
* @throws \Amp\Postgres\FailureException If connecting fails.
* @throws \Error If neither ext-pgsql or pecl-pq is loaded.
*/
function connect(string $connectionString, int $timeout = null): Awaitable {
function connect(string $connectionString, int $timeout = null): Promise {
if (\extension_loaded("pq")) {
return PqConnection::connect($connectionString, $timeout);
}

18
test/AbstractConnectionTest.php
View File

@ -180,8 +180,8 @@ abstract class AbstractConnectionTest extends \PHPUnit_Framework_TestCase {
}
public function testSimultaneousQueryAndPrepare() {
$awaitables = [];
$awaitables[] = new Coroutine((function () {
$promises = [];
$promises[] = new Coroutine((function () {
/** @var \Amp\Postgres\TupleResult $result */
$result = yield $this->connection->query("SELECT * FROM test");
@ -194,7 +194,7 @@ abstract class AbstractConnectionTest extends \PHPUnit_Framework_TestCase {
}
})());
$awaitables[] = new Coroutine((function () {
$promises[] = new Coroutine((function () {
/** @var \Amp\Postgres\Statement $statement */
$statement = (yield $this->connection->prepare("SELECT * FROM test"));
@ -210,13 +210,13 @@ abstract class AbstractConnectionTest extends \PHPUnit_Framework_TestCase {
}
})());
\Amp\execute(function () use ($awaitables) {
yield \Amp\all($awaitables);
\Amp\execute(function () use ($promises) {
yield \Amp\all($promises);
}, Loop::get());
}
public function testSimultaneousPrepareAndExecute() {
$awaitables[] = new Coroutine((function () {
$promises[] = new Coroutine((function () {
/** @var \Amp\Postgres\Statement $statement */
$statement = yield $this->connection->prepare("SELECT * FROM test");
@ -232,7 +232,7 @@ abstract class AbstractConnectionTest extends \PHPUnit_Framework_TestCase {
}
})());
$awaitables[] = new Coroutine((function () {
$promises[] = new Coroutine((function () {
/** @var \Amp\Postgres\TupleResult $result */
$result = yield $this->connection->execute("SELECT * FROM test");
@ -245,8 +245,8 @@ abstract class AbstractConnectionTest extends \PHPUnit_Framework_TestCase {
}
})());
\Amp\execute(function () use ($awaitables) {
yield \Amp\all($awaitables);
\Amp\execute(function () use ($promises) {
yield \Amp\all($promises);
}, Loop::get());
}

10
test/AbstractPoolTest.php
View File

@ -103,10 +103,10 @@ abstract class AbstractPoolTest extends \PHPUnit_Framework_TestCase {
\Amp\execute(function () Use ($count, $rounds, $pool, $method, $params) {
$awaitables = [];
$promises = [];
for ($i = 0; $i < $count * $rounds; ++$i) {
$awaitables[] = $pool->{$method}(...$params);
$promises[] = $pool->{$method}(...$params);
}
});
}
@ -169,14 +169,14 @@ abstract class AbstractPoolTest extends \PHPUnit_Framework_TestCase {
$pool = $this->createPool($connections);
\Amp\execute(function () use ($count, $rounds, $pool) {
$awaitables = [];
$promises = [];
for ($i = 0; $i < $count * $rounds; ++$i) {
$awaitables[] = $pool->transaction(Transaction::COMMITTED);
$promises[] = $pool->transaction(Transaction::COMMITTED);
}
yield \Amp\all(\Amp\map(function (Transaction $transaction) {
return $transaction->rollback();
}, $awaitables));
}, $promises));
});
}
}

4
test/ConnectionPoolTest.php
View File

@ -4,7 +4,7 @@ namespace Amp\Postgres\Test;
use Amp\Postgres\ConnectionPool;
use Amp\Success;
use Interop\Async\Awaitable;
use Interop\Async\Promise;
class ConnectionPoolTest extends AbstractPoolTest {
/**
@ -19,7 +19,7 @@ class ConnectionPoolTest extends AbstractPoolTest {
->getMock();
$mock->method('createConnection')
->will($this->returnCallback(function () use ($connections): Awaitable {
->will($this->returnCallback(function () use ($connections): Promise {
static $count = 0;
return new Success($connections[$count++]);
}));