Initial commit

.gitattributes

@@ -0,0 +1,6 @@
+example export-ignore
+test export-ignore
+.gitattributes export-ignore
+.gitignore export-ignore
+.travis.yml export-ignore
+phpunit.xml.dist export-ignore

.gitignore

@@ -0,0 +1,4 @@
+build
+composer.lock
+phpunit.xml
+vendor

.travis.yml

@@ -0,0 +1,39 @@
+sudo: false
+
+language: php
+
+php:
+  - 7.0
+  - 7.1
+  - nightly
+
+matrix:
+  allow_failures:
+    - php: 7.1
+    - php: nightly
+  fast_finish: true
+
+services:
+  - postgresql
+
+install:
+  - git clone https://github.com/m6w6/ext-pq;
+    pushd ext-pq;
+    phpize;
+    ./configure;
+    make;
+    make install;
+    popd;
+    echo "extension=pq.so" >> "$(php -r 'echo php_ini_loaded_file();')";
+  - composer self-update
+  - composer install --no-interaction --prefer-source
+
+before_script:
+  - psql -c 'CREATE DATABASE test;' -U postgres
+
+script:
+  - vendor/bin/phpunit --coverage-text --coverage-clover build/logs/clover.xml
+
+after_script:
+  - composer require satooshi/php-coveralls dev-master
+  - vendor/bin/coveralls -v --exclude-no-stmt

LICENSE

@@ -0,0 +1,21 @@
+The MIT License (MIT)
+
+Copyright (c) 2016 amphp
+
+Permission is hereby granted, free of charge, to any person obtaining a copy
+of this software and associated documentation files (the "Software"), to deal
+in the Software without restriction, including without limitation the rights
+to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+copies of the Software, and to permit persons to whom the Software is
+furnished to do so, subject to the following conditions:
+
+The above copyright notice and this permission notice shall be included in all
+copies or substantial portions of the Software.
+
+THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+SOFTWARE.

README.md

@@ -0,0 +1,61 @@
+# PostgreSQL Client for Amp
+
+This library is a component for [Amp](https://github.com/amphp/amp) that provides an asynchronous client for PostgreSQL.
+
+[![Build Status](https://img.shields.io/travis/amphp/postgres/master.svg?style=flat-square)](https://travis-ci.org/amphp/postgres)
+[![Coverage Status](https://img.shields.io/coveralls/amphp/postgres/master.svg?style=flat-square)](https://coveralls.io/r/amphp/postgres)
+[![Semantic Version](https://img.shields.io/github/release/amphp/postgres.svg?style=flat-square)](http://semver.org)
+[![MIT License](https://img.shields.io/packagist/l/amphp/postgres.svg?style=flat-square)](LICENSE)
+[![@amphp on Twitter](https://img.shields.io/badge/twitter-%40asyncphp-5189c7.svg?style=flat-square)](https://twitter.com/asyncphp)
+
+##### Requirements
+
+- PHP 7
+
+##### Installation
+
+The recommended way to install is with the [Composer](http://getcomposer.org/) package manager. (See the [Composer installation guide](https://getcomposer.org/doc/00-intro.md) for information on installing and using Composer.)
+
+Run the following command to use this library in your project: 
+
+```bash
+composer require amphp/postgres
+```
+
+You can also manually edit `composer.json` to add this library as a project requirement.
+
+```js
+// composer.json
+{
+    "require": {
+        "amphp/postgres": "^0.1"
+    }
+}
+```
+
+#### Example
+
+```php
+#!/usr/bin/env php
+<?php
+
+require __DIR__ . '/vendor/autoload.php';
+
+use Amp\Postgres;
+
+Amp\execute(function () {
+    /** @var \Amp\Postgres\Connection $connection */
+    $connection = yield Postgres\connect('host=localhost user=postgres dbname=test');
+
+    /** @var \Amp\Postgres\Statement $statement */
+    $statement = yield $connection->prepare('SELECT * FROM test WHERE id=$1');
+
+    /** @var \Amp\Postgres\TupleResult $result */
+    $result = yield $statement->execute(1337);
+
+    while (yield $result->next()) {
+        $row = $result->getCurrent();
+        // $row is an array (map) of column values. e.g.: $row['column_name']
+    }
+});
+```

composer.json

@@ -0,0 +1,43 @@
+{
+    "name": "amphp/postgres",
+    "description": "Asynchronous PostgreSQL client for Amp.",
+    "keywords": [
+        "database",
+        "db",
+        "postgresql",
+        "postgre",
+        "pgsql",
+        "asynchronous",
+        "async"
+    ],
+    "homepage": "http://amphp.org",
+    "license": "MIT",
+    "authors": [
+        {
+            "name": "Aaron Piotrowski",
+            "email": "aaron@trowski.com"
+        }
+    ],
+    "require": {
+        "amphp/amp": "dev-master as 2.0",
+        "async-interop/event-loop-implementation": "^0.3"
+    },
+    "require-dev": {
+        "amphp/loop": "dev-master",
+        "phpunit/phpunit": "^5.0"
+    },
+    "minimum-stability": "dev",
+    "autoload": {
+        "psr-4": {
+            "Amp\\Postgres\\": "lib"
+        },
+        "files": [
+            "lib/functions.php"
+        ]
+    },
+    "autoload-dev": {
+        "psr-4": {
+            "Amp\\Postgres\\Test\\": "test"
+        }
+    }
+}

example/test.php

@@ -0,0 +1,22 @@
+#!/usr/bin/env php
+<?php
+
+require dirname(__DIR__) . '/vendor/autoload.php';
+
+use Amp\Postgres;
+
+Amp\execute(function () {
+    /** @var \Amp\Postgres\Connection $connection */
+    $connection = yield Postgres\connect('host=localhost user=postgres');
+    
+    /** @var \Amp\Postgres\Statement $statement */
+    $statement = yield $connection->prepare('SHOW ALL');
+    
+    /** @var \Amp\Postgres\TupleResult $result */
+    $result = yield $statement->execute();
+    
+    while (yield $result->next()) {
+        $row = $result->getCurrent();
+        \printf("%-35s = %s (%s)\n", $row['name'], $row['setting'], $row['description']);
+    }
+});

lib/AbstractConnection.php

@@ -0,0 +1,113 @@
+<?php declare(strict_types = 1);
+
+namespace Amp\Postgres;
+
+use Amp\{ CallableMaker, Coroutine, Deferred, function pipe };
+use Interop\Async\Awaitable;
+
+abstract class AbstractConnection implements Connection {
+    use CallableMaker;
+    
+    /** @var \Amp\Postgres\PqConnection */
+    private $executor;
+    
+    /** @var \Amp\Deferred|null */
+    private $busy;
+    
+    /** @var callable */
+    private $release;
+    
+    /**
+     * @param string $connectionString
+     * @param int $timeout Timeout until the connection attempt fails.
+     *
+     * @return \Interop\Async\Awaitable<\Amp\Postgres\Connection>
+     */
+    abstract public static function connect(string $connectionString, int $timeout = null): Awaitable;
+    
+    /**
+     * @param $executor;
+     */
+    public function __construct(Executor $executor) {
+        $this->executor = $executor;
+        $this->release = $this->callableFromInstanceMethod("release");
+    }
+
+    /**
+     * @param callable $method Method to execute.
+     * @param mixed ...$args Arguments to pass to function.
+     *
+     * @return \Generator
+     *
+     * @resolve resource
+     *
+     * @throws \Amp\Postgres\FailureException
+     */
+    private function send(callable $method, ...$args): \Generator {
+        while ($this->busy !== null) {
+            yield $this->busy->getAwaitable();
+        }
+        
+        return $method(...$args);
+    }
+    
+    private function release() {
+        $busy = $this->busy;
+        $this->busy = null;
+        $busy->resolve();
+    }
+    
+    /**
+     * {@inheritdoc}
+     */
+    public function query(string $sql): Awaitable {
+        return new Coroutine($this->send([$this->executor, "query"], $sql));
+    }
+
+    /**
+     * {@inheritdoc}
+     */
+    public function execute(string $sql, ...$params): Awaitable {
+        return new Coroutine($this->send([$this->executor, "execute"], $sql, ...$params));
+    }
+
+    /**
+     * {@inheritdoc}
+     */
+    public function prepare(string $sql): Awaitable {
+        return new Coroutine($this->send([$this->executor, "prepare"], $sql, $sql));
+    }
+
+    /**
+     * {@inheritdoc}
+     */
+    public function transaction(int $isolation = Transaction::COMMITTED): Awaitable {
+        switch ($isolation) {
+            case Transaction::UNCOMMITTED:
+                $awaitable = $this->query("BEGIN TRANSACTION ISOLATION LEVEL READ UNCOMMITTED");
+                break;
+        
+            case Transaction::COMMITTED:
+                $awaitable = $this->query("BEGIN TRANSACTION ISOLATION LEVEL READ COMMITTED");
+                break;
+        
+            case Transaction::REPEATABLE:
+                $awaitable = $this->query("BEGIN TRANSACTION ISOLATION LEVEL REPEATABLE READ");
+                break;
+        
+            case Transaction::SERIALIZABLE:
+                $awaitable = $this->query("BEGIN TRANSACTION ISOLATION LEVEL SERIALIZABLE");
+                break;
+        
+            default:
+                throw new \Error("Invalid transaction type");
+        }
+    
+        return pipe($awaitable, function (CommandResult $result) use ($isolation) {
+            $this->busy = new Deferred;
+            $transaction = new Transaction($this->executor, $isolation);
+            $transaction->onComplete($this->release);
+            return $transaction;
+        });
+    }
+}

lib/AbstractPool.php

@@ -0,0 +1,217 @@
+<?php declare(strict_types = 1);
+
+namespace Amp\Postgres;
+
+use Amp\{ Coroutine, Deferred };
+use Interop\Async\Awaitable;
+
+abstract class AbstractPool implements Pool {
+    /** @var \SplQueue */
+    private $idle;
+
+    /** @var \SplQueue */
+    private $busy;
+
+    /** @var \SplObjectStorage */
+    private $connections;
+
+    /** @var \Amp\Deferred|\Interop\Async\Awaitable|null */
+    private $awaitable;
+
+    /**
+     * @return \Interop\Async\Awaitable<\Amp\Postgres\Connection>
+     *
+     * @throws \Amp\Postgres\FailureException
+     */
+    abstract protected function createConnection(): Awaitable;
+
+    public function __construct() {
+        $this->connections = new \SplObjectStorage();
+        $this->idle = new \SplQueue();
+        $this->busy = new \SplQueue();
+    }
+
+    /**
+     * {@inheritdoc}
+     */
+    public function getConnectionCount(): int {
+        return $this->connections->count();
+    }
+
+    /**
+     * {@inheritdoc}
+     */
+    public function getIdleConnectionCount(): int {
+        return $this->idle->count();
+    }
+
+    /**
+     * @param \Amp\Postgres\Connection $connection
+     */
+    protected function addConnection(Connection $connection) {
+        if (isset($this->connections[$connection])) {
+            return;
+        }
+
+        $this->connections->attach($connection);
+        $this->idle->push($connection);
+    }
+
+    /**
+     * @coroutine
+     *
+     * @return \Generator
+     *
+     * @resolve \Amp\Postgres\Connection
+     */
+    private function pop(): \Generator {
+        while (null !== $this->awaitable) {
+            try {
+                yield $this->awaitable; // Prevent simultaneous connection creation.
+            } catch (\Throwable $exception) {
+                // Ignore failure or cancellation of other operations.
+            }
+        }
+
+        if ($this->idle->isEmpty()) {
+            try {
+                if ($this->connections->count() >= $this->getMaxConnections()) {
+                    // All possible connections busy, so wait until one becomes available.
+                    $this->awaitable = new Deferred;
+                    yield $this->awaitable;
+                } else {
+                    // Max connection count has not been reached, so open another connection.
+                    $this->awaitable = $this->createConnection();
+                    $this->addConnection(yield $this->awaitable);
+                }
+            } finally {
+                $this->awaitable = null;
+            }
+        }
+
+        // Shift a connection off the idle queue.
+        return $this->idle->shift();
+    }
+
+    /**
+     * @param \Amp\Postgres\Connection $connection
+     *
+     * @throws \Error If the connection is not part of this pool.
+     */
+    private function push(Connection $connection) {
+        if (!isset($this->connections[$connection])) {
+            throw new \Error('Connection is not part of this pool');
+        }
+
+        $this->idle->push($connection);
+
+        if ($this->awaitable instanceof Deferred) {
+            $this->awaitable->resolve($connection);
+        }
+    }
+    
+    /**
+     * {@inheritdoc}
+     */
+    public function query(string $sql): Awaitable {
+        return new Coroutine($this->doQuery($sql));
+    }
+    
+    private function doQuery(string $sql): \Generator {
+        /** @var \Amp\Postgres\Connection $connection */
+        $connection = yield from $this->pop();
+
+        try {
+            $result = yield $connection->query($sql);
+        } catch (\Throwable $exception) {
+            $this->push($connection);
+            throw $exception;
+        }
+        
+        if ($result instanceof Operation) {
+            $result->onComplete(function () use ($connection) {
+                $this->push($connection);
+            });
+        } else {
+            $this->push($connection);
+        }
+
+        return $result;
+    }
+
+    /**
+     * {@inheritdoc}
+     */
+    public function execute(string $sql, ...$params): Awaitable {
+        return new Coroutine($this->doExecute($sql, $params));
+    }
+    
+    private function doExecute(string $sql, array $params): \Generator {
+        /** @var \Amp\Postgres\Connection $connection */
+        $connection = yield from $this->pop();
+    
+        try {
+            $result = yield $connection->execute($sql, ...$params);
+        } catch (\Throwable $exception) {
+            $this->push($connection);
+            throw $exception;
+        }
+    
+        if ($result instanceof Operation) {
+            $result->onComplete(function () use ($connection) {
+                $this->push($connection);
+            });
+        } else {
+            $this->push($connection);
+        }
+
+        return $result;
+    }
+
+    /**
+     * {@inheritdoc}
+     */
+    public function prepare(string $sql): Awaitable {
+        return new Coroutine($this->doPrepare($sql));
+    }
+    
+    private function doPrepare(string $sql): \Generator {
+        /** @var \Amp\Postgres\Connection $connection */
+        $connection = yield from $this->pop();
+
+        try {
+            /** @var \Amp\Postgres\Statement $statement */
+            $statement = yield $connection->prepare($sql);
+        } finally {
+            $this->push($connection);
+        }
+        
+        return $statement;
+    }
+
+    /**
+     * {@inheritdoc}
+     */
+    public function transaction(int $isolation = Transaction::COMMITTED): Awaitable {
+        return new Coroutine($this->doTransaction($isolation));
+    }
+    
+    private function doTransaction(int $isolation = Transaction::COMMITTED): \Generator {
+        /** @var \Amp\Postgres\Connection $connection */
+        $connection = yield from $this->pop();
+
+        try {
+            /** @var \Amp\Postgres\Transaction $transaction */
+            $transaction = yield $connection->transaction($isolation);
+        } catch (\Throwable $exception) {
+            $this->push($connection);
+            throw $exception;
+        }
+        
+        $transaction->onComplete(function () use ($connection) {
+            $this->push($connection);
+        });
+        
+        return $transaction;
+    }
+}

lib/AggregatePool.php

@@ -0,0 +1,34 @@
+<?php declare(strict_types = 1);
+
+namespace Amp\Postgres;
+
+use Interop\Async\Awaitable;
+
+class AggregatePool extends AbstractPool {
+    /**
+     * @param \Amp\Postgres\Connection $connection
+     */
+    public function addConnection(Connection $connection) {
+        parent::addConnection($connection);
+    }
+
+    /**
+     * {@inheritdoc}
+     */
+    protected function createConnection(): Awaitable {
+        throw new PoolError("Creating connections is not available in an aggregate pool");
+    }
+
+    /**
+     * {@inheritdoc}
+     */
+    public function getMaxConnections(): int {
+        $count = $this->getConnectionCount();
+
+        if (!$count) {
+            throw new PoolError("No connections in aggregate pool");
+        }
+
+        return $count;
+    }
+}

lib/CommandResult.php

@@ -0,0 +1,12 @@
+<?php declare(strict_types = 1);
+
+namespace Amp\Postgres;
+
+interface CommandResult extends \Countable, Result {
+    /**
+     * Returns the number of rows affected by the query.
+     *
+     * @return int
+     */
+    public function affectedRows(): int;
+}

lib/Connection.php

@@ -0,0 +1,16 @@
+<?php declare(strict_types = 1);
+
+namespace Amp\Postgres;
+
+use Interop\Async\Awaitable;
+
+interface Connection extends Executor {
+    /**
+     * @param int $isolation
+     *
+     * @return \Interop\Async\Awaitable<\Amp\Postgres\Transaction>
+     *
+     * @throws \Amp\Postgres\FailureException
+     */
+    public function transaction(int $isolation = Transaction::COMMITTED): Awaitable;
+}

lib/ConnectionPool.php

@@ -0,0 +1,54 @@
+<?php declare(strict_types = 1);
+
+namespace Amp\Postgres;
+
+use Interop\Async\Awaitable;
+
+class ConnectionPool extends AbstractPool {
+    const DEFAULT_MAX_CONNECTIONS = 100;
+    const DEFAULT_CONNECT_TIMEOUT = 5000;
+
+    /** @var string */
+    private $connectionString;
+
+    /** @var int */
+    private $connectTimeout;
+
+    /** @var int */
+    private $maxConnections;
+
+    /**
+     * @param string $connectionString
+     * @param int $maxConnections
+     * @param int $connectTimeout
+     */
+    public function __construct(
+        string $connectionString,
+        int $maxConnections = self::DEFAULT_MAX_CONNECTIONS,
+        int $connectTimeout = self::DEFAULT_CONNECT_TIMEOUT
+    ) {
+        parent::__construct();
+
+        $this->connectionString = $connectionString;
+        $this->connectTimeout = $connectTimeout;
+
+        $this->maxConnections = $maxConnections;
+        if (1 > $this->maxConnections) {
+            $this->maxConnections = 1;
+        }
+    }
+
+    /**
+     * {@inheritdoc}
+     */
+    protected function createConnection(): Awaitable {
+        return connect($this->connectionString, $this->connectTimeout);
+    }
+
+    /**
+     * {@inheritdoc}
+     */
+    public function getMaxConnections(): int {
+        return $this->maxConnections;
+    }
+}

lib/Executor.php

@@ -0,0 +1,35 @@
+<?php declare(strict_types = 1);
+
+namespace Amp\Postgres;
+
+use Interop\Async\Awaitable;
+
+interface Executor {
+    /**
+     * @param string $sql
+     *
+     * @return \Interop\Async\Awaitable<\Amp\Postgres\Result>
+     *
+     * @throws \Amp\Postgres\FailureException
+     */
+    public function query(string $sql): Awaitable;
+
+    /**
+     * @param string $sql
+     * @param mixed ...$params
+     *
+     * @return \Interop\Async\Awaitable<\Amp\Postgres\Result>
+     *
+     * @throws \Amp\Postgres\FailureException
+     */
+    public function execute(string $sql, ...$params): Awaitable;
+
+    /**
+     * @param string $sql
+     *
+     * @return \Interop\Async\Awaitable<\Amp\Postgres\Statement>
+     *
+     * @throws \Amp\Postgres\FailureException
+     */
+    public function prepare(string $sql): Awaitable;
+}

lib/FailureException.php

@@ -0,0 +1,5 @@
+<?php declare(strict_types = 1);
+
+namespace Amp\Postgres;
+
+class FailureException extends \Exception {}

lib/Internal/Operation.php

@@ -0,0 +1,36 @@
+<?php declare(strict_types = 1);
+
+namespace Amp\Postgres\Internal;
+
+trait Operation {
+    /** @var bool */
+    private $complete = false;
+    
+    /** @var callable[] */
+    private $onComplete = [];
+    
+    public function __destruct() {
+        $this->complete();
+    }
+    
+    public function onComplete(callable $onComplete) {
+        if ($this->complete) {
+            $onComplete();
+            return;
+        }
+        
+        $this->onComplete[] = $onComplete;
+    }
+    
+    private function complete() {
+        if ($this->complete) {
+            return;
+        }
+        
+        $this->complete = true;
+        foreach ($this->onComplete as $callback) {
+            $callback();
+        }
+        $this->onComplete = null;
+    }
+}

lib/Operation.php

@@ -0,0 +1,10 @@
+<?php declare(strict_types = 1);
+
+namespace Amp\Postgres;
+
+interface Operation {
+    /**
+     * @param callable $onComplete Callback executed when the operation completes or the object is destroyed.
+     */
+    public function onComplete(callable $onComplete);
+}

lib/PgSqlCommandResult.php

@@ -0,0 +1,43 @@
+<?php declare(strict_types = 1);
+
+namespace Amp\Postgres;
+
+class PgSqlCommandResult implements CommandResult {
+    /** @var resource PostgreSQL result resource. */
+    private $handle;
+
+    /**
+     * @param resource $handle PostgreSQL result resource.
+     */
+    public function __construct($handle) {
+        $this->handle = $handle;
+    }
+
+    /**
+     * Frees the result resource.
+     */
+    public function __destruct() {
+        \pg_free_result($this->handle);
+    }
+
+    /**
+     * @return int Number of rows affected by the INSERT, UPDATE, or DELETE query.
+     */
+    public function affectedRows(): int {
+        return \pg_affected_rows($this->handle);
+    }
+
+    /**
+     * @return string
+     */
+    public function lastOid(): string {
+        return (string) \pg_last_oid($this->handle);
+    }
+
+    /**
+     * @return int
+     */
+    public function count(): int {
+        return $this->affectedRows();
+    }
+}

lib/PgSqlConnection.php

@@ -0,0 +1,96 @@
+<?php declare(strict_types = 1);
+
+namespace Amp\Postgres;
+
+use Amp\{ Deferred, TimeoutException };
+use Interop\Async\{ Awaitable, Loop };
+
+class PgSqlConnection extends AbstractConnection {
+    
+    /** @var \Amp\Postgres\PqConnection */
+    private $executor;
+    
+    /** @var \Amp\Deferred|null */
+    private $busy;
+    
+    /** @var callable */
+    private $release;
+    
+    /**
+     * @param string $connectionString
+     * @param int|null $timeout
+     *
+     * @return \Interop\Async\Awaitable<\Amp\Postgres\PgSqlConnection>
+     *
+     * @throws \Amp\Postgres\FailureException
+     */
+    public static function connect(string $connectionString, int $timeout = null): Awaitable {
+        if (!$connection = @\pg_connect($connectionString, \PGSQL_CONNECT_ASYNC | \PGSQL_CONNECT_FORCE_NEW)) {
+            throw new FailureException("Failed to create connection resource");
+        }
+    
+        if (\pg_connection_status($connection) === \PGSQL_CONNECTION_BAD) {
+            throw new FailureException(\pg_last_error($connection));
+        }
+    
+        if (!$socket = \pg_socket($connection)) {
+            throw new FailureException("Failed to access connection socket");
+        }
+    
+        $deferred = new Deferred;
+    
+        $callback = function ($watcher, $resource) use (&$poll, &$await, $connection, $deferred) {
+            try {
+                switch (\pg_connect_poll($connection)) {
+                    case \PGSQL_POLLING_READING:
+                        return; // Connection not ready, poll again.
+                
+                    case \PGSQL_POLLING_WRITING:
+                        return; // Still writing...
+                
+                    case \PGSQL_POLLING_FAILED:
+                        throw new FailureException("Could not connect to PostgreSQL server");
+                
+                    case \PGSQL_POLLING_OK:
+                        Loop::cancel($poll);
+                        Loop::cancel($await);
+                        $deferred->resolve(new self($connection, $resource));
+                        return;
+                }
+            } catch (\Throwable $exception) {
+                Loop::cancel($poll);
+                Loop::cancel($await);
+                \pg_close($connection);
+                $deferred->fail($exception);
+            }
+        };
+    
+        $poll = Loop::onReadable($socket, $callback);
+        $await = Loop::onWritable($socket, $callback);
+    
+        if ($timeout !== null) {
+            return \Amp\capture(
+                $deferred->getAwaitable(),
+                TimeoutException::class,
+                function (\Throwable $exception) use ($connection, $poll, $await) {
+                    Loop::cancel($poll);
+                    Loop::cancel($await);
+                    \pg_close($connection);
+                    throw $exception;
+                }
+            );
+        }
+    
+        return $deferred->getAwaitable();
+    }
+    
+    /**
+     * Connection constructor.
+     *
+     * @param resource $handle PostgreSQL connection handle.
+     * @param resource $socket PostgreSQL connection stream socket.
+     */
+    public function __construct($handle, $socket) {
+        parent::__construct(new PgSqlExecutor($handle, $socket));
+    }
+}

lib/PgSqlExecutor.php

@@ -0,0 +1,190 @@
+<?php declare(strict_types = 1);
+
+namespace Amp\Postgres;
+
+use Amp\{ CallableMaker, Coroutine, Deferred, function pipe };
+use Interop\Async\{ Awaitable, Loop };
+
+class PgSqlExecutor implements Executor {
+    use CallableMaker;
+    
+    /** @var resource PostgreSQL connection handle. */
+    private $handle;
+
+    /** @var \Amp\Deferred|null */
+    private $delayed;
+
+    /** @var string */
+    private $poll;
+
+    /** @var string */
+    private $await;
+
+    /** @var callable */
+    private $executeCallback;
+    
+    /** @var callable */
+    private $createResult;
+    
+    /**
+     * Connection constructor.
+     *
+     * @param resource $handle PostgreSQL connection handle.
+     * @param resource $socket PostgreSQL connection stream socket.
+     */
+    public function __construct($handle, $socket) {
+        $this->handle = $handle;
+
+        $deferred = &$this->delayed;
+        
+        $this->poll = Loop::onReadable($socket, static function ($watcher) use (&$deferred, $handle) {
+            if (!\pg_consume_input($handle)) {
+                Loop::disable($watcher);
+                $deferred->fail(new FailureException(\pg_last_error($handle)));
+                return;
+            }
+
+            if (!\pg_connection_busy($handle)) {
+                Loop::disable($watcher);
+                $deferred->resolve(\pg_get_result($handle));
+                return;
+            }
+
+            // Reading not done, listen again.
+        });
+
+        $this->await = Loop::onWritable($socket, static function ($watcher) use (&$deferred, $handle) {
+            $flush = \pg_flush($handle);
+            if (0 === $flush) {
+                return; // Not finished sending data, listen again.
+            }
+            
+            Loop::disable($watcher);
+
+            if ($flush === false) {
+                $deferred->fail(new FailureException(\pg_last_error($handle)));
+            }
+        });
+        
+        Loop::disable($this->poll);
+        Loop::disable($this->await);
+
+        $this->createResult = $this->callableFromInstanceMethod("createResult");
+        $this->executeCallback = $this->callableFromInstanceMethod("sendExecute");
+    }
+
+    /**
+     * Frees Io watchers from loop.
+     */
+    public function __destruct() {
+        if (\is_resource($this->handle)) {
+            \pg_close($this->handle);
+        }
+        
+        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
+     */
+    private function send(callable $function, ...$args): \Generator {
+        while ($this->delayed !== null) {
+            try {
+                yield $this->delayed->getAwaitable();
+            } catch (\Throwable $exception) {
+                // Ignore failure from another operation.
+            }
+        }
+
+        $result = $function($this->handle, ...$args);
+
+        if ($result === false) {
+            throw new FailureException(\pg_last_error($this->handle));
+        }
+
+        $this->delayed = new Deferred;
+
+        Loop::enable($this->poll);
+        if (0 === $result) {
+            Loop::enable($this->await);
+        }
+
+        try {
+            $result = yield $this->delayed->getAwaitable();
+        } finally {
+            $this->delayed = null;
+            Loop::disable($this->poll);
+            Loop::disable($this->await);
+        }
+
+        return $result;
+    }
+    
+    /**
+     * @param resource $result PostgreSQL result resource.
+     *
+     * @return \Amp\Postgres\Result
+     *
+     * @throws \Amp\Postgres\FailureException
+     * @throws \Amp\Postgres\QueryError
+     */
+    private function createResult($result): Result {
+        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:
+                throw new FailureException("Unknown result status");
+        }
+    }
+    
+    private function sendExecute(string $name, array $params): Awaitable {
+        return pipe(new Coroutine($this->send("pg_send_execute", $name, $params)), $this->createResult);
+    }
+
+    /**
+     * {@inheritdoc}
+     */
+    public function query(string $sql): Awaitable {
+        return pipe(new Coroutine($this->send("pg_send_query", $sql)), $this->createResult);
+    }
+
+    /**
+     * {@inheritdoc}
+     */
+    public function execute(string $sql, ...$params): Awaitable {
+        return pipe(new Coroutine($this->send("pg_send_query_params", $sql, $params)), $this->createResult);
+    }
+
+    /**
+     * {@inheritdoc}
+     */
+    public function prepare(string $sql): Awaitable {
+        return pipe(new Coroutine($this->send("pg_send_prepare", $sql, $sql)), function () use ($sql) {
+            return new PgSqlStatement($sql, $this->executeCallback);
+        });
+    }
+}

lib/PgSqlStatement.php

@@ -0,0 +1,40 @@
+<?php declare(strict_types = 1);
+
+namespace Amp\Postgres;
+
+use Interop\Async\Awaitable;
+
+class PgSqlStatement implements Statement {
+    /** @var string */
+    private $sql;
+
+    /** @var callable */
+    private $execute;
+
+    /**
+     * @param string $sql
+     * @param callable $execute
+     */
+    public function __construct(string $sql, callable $execute) {
+        $this->sql = $sql;
+        $this->execute = $execute;
+    }
+
+    /**
+     * @return string
+     */
+    public function getQuery(): string {
+        return $this->sql;
+    }
+
+    /**
+     * @param mixed ...$params
+     *
+     * @return \Interop\Async\Awaitable<\Amp\Postgres\Result>
+     *
+     * @throws \Amp\Postgres\FailureException If executing the statement fails.
+     */
+    public function execute(...$params): Awaitable {
+        return ($this->execute)($this->sql, $params);
+    }
+}

lib/PgSqlTupleResult.php

@@ -0,0 +1,129 @@
+<?php declare(strict_types = 1);
+
+namespace Amp\Postgres;
+
+use Amp\Emitter;
+
+class PgSqlTupleResult extends TupleResult implements \Countable {
+    /** @var resource PostgreSQL result resource. */
+    private $handle;
+
+    /**
+     * @param resource $handle PostgreSQL result resource.
+     */
+    public function __construct($handle) {
+        $this->handle = $handle;
+        parent::__construct(new Emitter(static function (callable $emit) use ($handle) {
+            $count = \pg_num_rows($handle);
+            for ($i = 0; $i < $count; ++$i) {
+                $result = \pg_fetch_assoc($handle);
+                if ($result === false) {
+                    throw new FailureException(\pg_result_error($handle));
+                }
+                yield $emit($result);
+            }
+            return $count;
+        }));
+    }
+    
+    /**
+     * Frees the result resource.
+     */
+    public function __destruct() {
+        \pg_free_result($this->handle);
+    }
+    
+    /**
+     * @return int Number of rows in the result set.
+     */
+    public function numRows(): int {
+        return \pg_num_rows($this->handle);
+    }
+
+    /**
+     * @return int Number of fields in each row.
+     */
+    public function numFields(): int {
+        return \pg_num_fields($this->handle);
+    }
+
+    /**
+     * @param int $fieldNum
+     *
+     * @return string Column name at index $fieldNum
+     *
+     * @throws \Error If the field number does not exist in the result.
+     */
+    public function fieldName(int $fieldNum): string {
+        return \pg_field_name($this->handle, $this->filterNameOrNum($fieldNum));
+    }
+
+    /**
+     * @param string $fieldName
+     *
+     * @return int Index of field with given name.
+     *
+     * @throws \Error If the field name does not exist in the result.
+     */
+    public function fieldNum(string $fieldName): int {
+        $result = \pg_field_num($this->handle, $fieldName);
+
+        if (-1 === $result) {
+            throw new \Error(\sprintf('No field with name "%s" in result', $fieldName));
+        }
+
+        return $result;
+    }
+
+    /**
+     * @param int|string $fieldNameOrNum Field name or index.
+     *
+     * @return string Name of the field type.
+     *
+     * @throws \Error If the field number does not exist in the result.
+     */
+    public function fieldType($fieldNameOrNum): string {
+        return \pg_field_type($this->handle, $this->filterNameOrNum($fieldNameOrNum));
+    }
+
+    /**
+     * @param int|string $fieldNameOrNum Field name or index.
+     *
+     * @return int Storage required for field. -1 indicates a variable length field.
+     *
+     * @throws \Error If the field number does not exist in the result.
+     */
+    public function fieldSize($fieldNameOrNum): int {
+        return \pg_field_size($this->handle, $this->filterNameOrNum($fieldNameOrNum));
+    }
+
+    /**
+     * @return int Number of rows in the result set.
+     */
+    public function count(): int {
+        return $this->numRows();
+    }
+
+    /**
+     * @param int|string $fieldNameOrNum Field name or index.
+     *
+     * @return int Field index.
+     *
+     * @throws \Error
+     */
+    private function filterNameOrNum($fieldNameOrNum): int {
+        if (\is_string($fieldNameOrNum)) {
+            return $this->fieldNum($fieldNameOrNum);
+        }
+
+        if (!\is_int($fieldNameOrNum)) {
+            throw new \Error('Must provide a string name or integer field number');
+        }
+
+        if (0 > $fieldNameOrNum || $this->numFields() <= $fieldNameOrNum) {
+            throw new \Error(\sprintf('No field with index %d in result', $fieldNameOrNum));
+        }
+
+        return $fieldNameOrNum;
+    }
+}

lib/Pool.php

@@ -0,0 +1,20 @@
+<?php declare(strict_types = 1);
+
+namespace Amp\Postgres;
+
+interface Pool extends Connection {
+    /**
+     * @return int Current number of connections in the pool.
+     */
+    public function getConnectionCount(): int;
+
+    /**
+     * @return int Current number of idle connections in the pool.
+     */
+    public function getIdleConnectionCount(): int;
+
+    /**
+     * @return int Maximum number of connections.
+     */
+    public function getMaxConnections(): int;
+}

lib/PoolError.php

@@ -0,0 +1,5 @@
+<?php declare(strict_types = 1);
+
+namespace Amp\Postgres;
+
+class PoolError extends \Error {}

lib/PqBufferedResult.php

@@ -0,0 +1,36 @@
+<?php declare(strict_types = 1);
+
+namespace Amp\Postgres;
+
+use Amp\Emitter;
+use pq;
+
+class PqBufferedResult extends TupleResult implements \Countable {
+    /** @var \pq\Result */
+    private $result;
+
+    /**
+     * @param pq\Result $result PostgreSQL result object.
+     */
+    public function __construct(pq\Result $result) {
+        $this->result = $result;
+        parent::__construct(new Emitter(static function (callable $emit) use ($result) {
+            for ($count = 0; $row = $result->fetchRow(pq\Result::FETCH_ASSOC); ++$count) {
+                yield $emit($row);
+            }
+            return $count;
+        }));
+    }
+    
+    public function numRows(): int {
+        return $this->result->numRows;
+    }
+    
+    public function numFields(): int {
+        return $this->result->numCols;
+    }
+    
+    public function count(): int {
+        return $this->numRows();
+    }
+}

lib/PqCommandResult.php

@@ -0,0 +1,30 @@
+<?php declare(strict_types = 1);
+
+namespace Amp\Postgres;
+use pq;
+
+class PqCommandResult implements CommandResult {
+    /** @var \pq\Result PostgreSQL result object. */
+    private $result;
+
+    /**
+     * @param \pq\Result $result PostgreSQL result object.
+     */
+    public function __construct(pq\Result $result) {
+        $this->result = $result;
+    }
+
+    /**
+     * @return int Number of rows affected by the INSERT, UPDATE, or DELETE query.
+     */
+    public function affectedRows(): int {
+        return $this->result->affectedRows;
+    }
+
+    /**
+     * @return int
+     */
+    public function count() {
+        return $this->affectedRows();
+    }
+}

lib/PqConnection.php

@@ -0,0 +1,91 @@
+<?php declare(strict_types = 1);
+
+namespace Amp\Postgres;
+
+use Amp\{ Deferred, TimeoutException };
+use Interop\Async\{ Awaitable, Loop };
+use pq;
+
+class PqConnection extends AbstractConnection {
+    /** @var \Amp\Postgres\PqConnection */
+    private $executor;
+    
+    /** @var \Amp\Deferred|null */
+    private $busy;
+    
+    /** @var callable */
+    private $release;
+    
+    /**
+     * @param string $connectionString
+     * @param int|null $timeout
+     *
+     * @return \Interop\Async\Awaitable<\Amp\Postgres\PgSqlConnection>
+     *
+     * @throws \Amp\Postgres\FailureException
+     */
+    public static function connect(string $connectionString, int $timeout = null): Awaitable {
+        try {
+            $connection = new pq\Connection($connectionString, pq\Connection::ASYNC);
+        } catch (pq\Exception $exception) {
+            throw new FailureException("Could not connect to PostgresSQL server", 0, $exception);
+        }
+        $connection->resetAsync();
+        $connection->nonblocking = true;
+        $connection->unbuffered = true;
+        
+        $deferred = new Deferred;
+    
+        $callback = function ($watcher, $resource) use (&$poll, &$await, $connection, $deferred) {
+            try {
+                switch ($connection->poll()) {
+                    case pq\Connection::POLLING_READING:
+                        return; // Connection not ready, poll again.
+    
+                    case pq\Connection::POLLING_WRITING:
+                        return; // Still writing...
+    
+                    case pq\Connection::POLLING_FAILED:
+                        throw new FailureException("Could not connect to PostgreSQL server");
+                    
+                    case pq\Connection::POLLING_OK:
+                    case \PGSQL_POLLING_OK:
+                        Loop::cancel($poll);
+                        Loop::cancel($await);
+                        $deferred->resolve(new self($connection));
+                        return;
+                }
+            } catch (\Throwable $exception) {
+                Loop::cancel($poll);
+                Loop::cancel($await);
+                $deferred->fail($exception);
+            }
+        };
+    
+        $poll = Loop::onReadable($connection->socket, $callback);
+        $await = Loop::onWritable($connection->socket, $callback);
+    
+        if ($timeout !== null) {
+            return \Amp\capture(
+                $deferred->getAwaitable(),
+                TimeoutException::class,
+                function (\Throwable $exception) use ($connection, $poll, $await) {
+                    Loop::cancel($poll);
+                    Loop::cancel($await);
+                    throw $exception;
+                }
+            );
+        }
+    
+        return $deferred->getAwaitable();
+    }
+    
+    /**
+     * Connection constructor.
+     *
+     * @param \pq\Connection $handle
+     */
+    public function __construct(pq\Connection $handle) {
+        parent::__construct(new PqExecutor($handle));
+    }
+}

lib/PqExecutor.php

@@ -0,0 +1,216 @@
+<?php declare(strict_types = 1);
+
+namespace Amp\Postgres;
+
+use Amp\{ CallableMaker, Coroutine, Deferred, function pipe };
+use Interop\Async\{ Awaitable, Loop };
+use pq;
+
+class PqExecutor implements Executor {
+    use CallableMaker;
+    
+    /** @var \pq\Connection PostgreSQL connection object. */
+    private $handle;
+
+    /** @var \Amp\Deferred|null */
+    private $delayed;
+    
+    /** @var \Amp\Deferred */
+    private $busy;
+
+    /** @var string */
+    private $poll;
+
+    /** @var string */
+    private $await;
+
+    /** @var callable */
+    private $send;
+    
+    /** @var callable */
+    private $fetch;
+    
+    /** @var callable */
+    private $release;
+    
+    /**
+     * Connection constructor.
+     *
+     * @param \pq\Connection $handle
+     */
+    public function __construct(pq\Connection $handle) {
+        $this->handle = $handle;
+
+        $deferred = &$this->delayed;
+        
+        $this->poll = Loop::onReadable($this->handle->socket, static function ($watcher) use (&$deferred, $handle) {
+            if ($handle->poll() === pq\Connection::POLLING_FAILED) {
+                $deferred->fail(new FailureException($handle->errorMessage));
+                return;
+            }
+
+            if (!$handle->busy) {
+                $deferred->resolve($handle->getResult());
+                return;
+            }
+
+            // Reading not done, listen again.
+        });
+
+        $this->await = Loop::onWritable($this->handle->socket, static function ($watcher) use (&$deferred, $handle) {
+            if (!$handle->flush()) {
+                return; // Not finished sending data, listen again.
+            }
+            
+            Loop::disable($watcher);
+        });
+        
+        Loop::disable($this->poll);
+        Loop::disable($this->await);
+
+        $this->send = $this->callableFromInstanceMethod("send");
+        $this->fetch = $this->callableFromInstanceMethod("fetch");
+        $this->release = $this->callableFromInstanceMethod("release");
+    }
+
+    /**
+     * Frees Io watchers from loop.
+     */
+    public function __destruct() {
+        Loop::cancel($this->poll);
+        Loop::cancel($this->await);
+    }
+
+    /**
+     * @param callable $method Method to execute.
+     * @param mixed ...$args Arguments to pass to function.
+     *
+     * @return \Generator
+     *
+     * @resolve resource
+     *
+     * @throws \Amp\Postgres\FailureException
+     */
+    private function send(callable $method, ...$args): \Generator {
+        while ($this->busy !== null) {
+            yield $this->busy->getAwaitable();
+        }
+        
+        $this->busy = new Deferred;
+
+        try {
+            try {
+                $handle = $method(...$args);
+            } catch (pg\Exception $exception) {
+                throw new FailureException($this->handle->errorMessage, 0, $exception);
+            }
+    
+            $this->delayed = new Deferred;
+    
+            Loop::enable($this->poll);
+            if (!$this->handle->flush()) {
+                Loop::enable($this->await);
+            }
+    
+            try {
+                $result = yield $this->delayed->getAwaitable();
+            } finally {
+                $this->delayed = null;
+                Loop::disable($this->poll);
+                Loop::disable($this->await);
+            }
+    
+            if ($handle instanceof pq\Statement) {
+                return new PqStatement($handle, $this->send);
+            }
+            
+            if (!$result instanceof pq\Result) {
+                throw new FailureException("Unknown query result");
+            }
+        } finally {
+            $this->release();
+        }
+    
+        switch ($result->status) {
+            case pq\Result::EMPTY_QUERY:
+                throw new QueryError("Empty query string");
+        
+            case pq\Result::COMMAND_OK:
+                return new PqCommandResult($result);
+        
+            case pq\Result::TUPLES_OK:
+                return new PqBufferedResult($result);
+        
+            CASE pq\Result::SINGLE_TUPLE:
+                $result = new PqUnbufferedResult($this->fetch, $result);
+                $result->onComplete($this->release);
+                $this->busy = new Deferred;
+                return $result;
+        
+            case pq\Result::NONFATAL_ERROR:
+            case pq\Result::FATAL_ERROR:
+                throw new QueryError($result->errorMessage);
+        
+            case pq\Result::BAD_RESPONSE:
+                throw new FailureException($result->errorMessage);
+        
+            default:
+                throw new FailureException("Unknown result status");
+        }
+    }
+    
+    private function fetch(): \Generator {
+        if (!$this->handle->busy) { // Results buffered.
+            $result = $this->handle->getResult();
+        } else {
+            $this->delayed = new Deferred;
+    
+            Loop::enable($this->poll);
+    
+            try {
+                $result = yield $this->delayed->getAwaitable();
+            } finally {
+                $this->delayed = null;
+                Loop::disable($this->poll);
+            }
+        }
+        
+        switch ($result->status) {
+            case pq\Result::TUPLES_OK: // No more rows in result set.
+                return null;
+            
+            case pq\Result::SINGLE_TUPLE:
+                return $result;
+            
+            default:
+                throw new FailureException($result->errorMessage);
+        }
+    }
+    
+    private function release() {
+        $busy = $this->busy;
+        $this->busy = null;
+        $busy->resolve();
+    }
+    
+    /**
+     * {@inheritdoc}
+     */
+    public function query(string $sql): Awaitable {
+        return new Coroutine($this->send([$this->handle, "execAsync"], $sql));
+    }
+
+    /**
+     * {@inheritdoc}
+     */
+    public function execute(string $sql, ...$params): Awaitable {
+        return new Coroutine($this->send([$this->handle, "execParamsAsync"], $sql, $params));
+    }
+
+    /**
+     * {@inheritdoc}
+     */
+    public function prepare(string $sql): Awaitable {
+        return new Coroutine($this->send([$this->handle, "prepareAsync"], $sql, $sql));
+    }
+}

lib/PqStatement.php

@@ -0,0 +1,46 @@
+<?php declare(strict_types = 1);
+
+namespace Amp\Postgres;
+
+use Amp\{ Coroutine, function rethrow };
+use Interop\Async\Awaitable;
+use pq;
+
+class PqStatement implements Statement {
+    /** @var \pq\Statement */
+    private $statement;
+
+    /** @var callable */
+    private $execute;
+
+    /**
+     * @param \pq\Statement $statement
+     * @param callable $execute
+     */
+    public function __construct(pq\Statement $statement, callable $execute) {
+        $this->statement = $statement;
+        $this->execute = $execute;
+    }
+    
+    public function __destruct() {
+        rethrow(new Coroutine(($this->execute)([$this->statement, "deallocateAsync"])));
+    }
+    
+    /**
+     * @return string
+     */
+    public function getQuery(): string {
+        return $this->statement->query;
+    }
+
+    /**
+     * @param mixed ...$params
+     *
+     * @return \Interop\Async\Awaitable<\Amp\Postgres\Result>
+     *
+     * @throws \Amp\Postgres\FailureException If executing the statement fails.
+     */
+    public function execute(...$params): Awaitable {
+        return new Coroutine(($this->execute)([$this->statement, "execAsync"], $params));
+    }
+}

lib/PqUnbufferedResult.php

@@ -0,0 +1,39 @@
+<?php declare(strict_types = 1);
+
+namespace Amp\Postgres;
+
+use Amp\{ Coroutine, Emitter };
+use pq;
+
+class PqUnbufferedResult extends TupleResult implements Operation {
+    use Internal\Operation;
+    
+    /** @var int */
+    private $numCols;
+    
+    /**
+     * @param callable(): \Generator $fetch Coroutine function to fetch next result row.
+     * @param \pq\Result $result PostgreSQL result object.
+     */
+    public function __construct(callable $fetch, pq\Result $result) {
+        $this->numCols = $result->numCols;
+        parent::__construct(new Emitter(function (callable $emit) use ($result, $fetch) {
+            $count = 0;
+            try {
+                do {
+                    $next = new Coroutine($fetch()); // Request next result before current is consumed.
+                    ++$count;
+                    yield $emit($result->fetchRow(pq\Result::FETCH_ASSOC));
+                    $result = yield $next;
+                } while ($result instanceof pq\Result);
+            } finally {
+                $this->complete();
+            }
+            return $count;
+        }));
+    }
+    
+    public function numFields(): int {
+        return $this->numCols;
+    }
+}

lib/QueryError.php

@@ -0,0 +1,5 @@
+<?php declare(strict_types = 1);
+
+namespace Amp\Postgres;
+
+class QueryError extends \Error {}

lib/Result.php

@@ -0,0 +1,5 @@
+<?php declare(strict_types = 1);
+
+namespace Amp\Postgres;
+
+interface Result {}

lib/Statement.php

@@ -0,0 +1,14 @@
+<?php declare(strict_types = 1);
+
+namespace Amp\Postgres;
+
+use Interop\Async\Awaitable;
+
+interface Statement {
+    /**
+     * @param mixed ...$params
+     *
+     * @return \Interop\Async\Awaitable<\Amp\Postgres\Result>
+     */
+    public function execute(...$params): Awaitable;
+}

lib/Transaction.php

@@ -0,0 +1,185 @@
+<?php declare(strict_types = 1);
+
+namespace Amp\Postgres;
+
+use Amp\Coroutine;
+use Interop\Async\Awaitable;
+
+class Transaction implements Executor, Operation {
+    use Internal\Operation;
+    
+    const UNCOMMITTED  = 0;
+    const COMMITTED    = 1;
+    const REPEATABLE   = 2;
+    const SERIALIZABLE = 4;
+
+    /** @var \Amp\Postgres\Executor */
+    private $executor;
+
+    /** @var int */
+    private $isolation;
+
+    /**
+     * @param \Amp\Postgres\Executor $executor
+     * @param int $isolation
+     *
+     * @throws \Error If the isolation level is invalid.
+     */
+    public function __construct(Executor $executor, int $isolation = self::COMMITTED) {
+        switch ($isolation) {
+            case self::UNCOMMITTED:
+            case self::COMMITTED:
+            case self::REPEATABLE:
+            case self::SERIALIZABLE:
+                $this->isolation = $isolation;
+                break;
+
+            default:
+                throw new \Error("Isolation must be a valid transaction isolation level");
+        }
+
+        $this->executor = $executor;
+    }
+    
+    /**
+     * @return bool
+     */
+    public function isActive(): bool {
+        return $this->executor !== null;
+    }
+
+    /**
+     * @return int
+     */
+    public function getIsolationLevel(): int {
+        return $this->isolation;
+    }
+    
+    /**
+     * {@inheritdoc}
+     */
+    public function query(string $sql): Awaitable {
+        if ($this->executor === null) {
+            throw new TransactionError("The transaction has been committed or rolled back");
+        }
+
+        return $this->executor->query($sql);
+    }
+
+    /**
+     * {@inheritdoc}
+     */
+    public function prepare(string $sql): Awaitable {
+        if ($this->executor === null) {
+            throw new TransactionError("The transaction has been committed or rolled back");
+        }
+
+        return $this->executor->prepare($sql);
+    }
+
+    /**
+     * {@inheritdoc}
+     */
+    public function execute(string $sql, ...$params): Awaitable {
+        if ($this->executor === null) {
+            throw new TransactionError("The transaction has been committed or rolled back");
+        }
+
+        return $this->executor->execute($sql, ...$params);
+    }
+
+    /**
+     * Commits the transaction and makes it inactive.
+     *
+     * @return \Interop\Async\Awaitable<\Amp\Postgres\CommandResult>
+     *
+     * @throws \Amp\Postgres\TransactionError
+     */
+    public function commit(): Awaitable {
+        return new Coroutine($this->doCommit());
+    }
+    
+    private function doCommit(): \Generator {
+        if ($this->executor === null) {
+            throw new TransactionError("The transaction has been committed or rolled back");
+        }
+
+        $executor = $this->executor;
+        $this->executor = null;
+
+        try {
+            $result = yield $executor->query("COMMIT");
+        } finally {
+            $this->complete();
+        }
+
+        return $result;
+    }
+
+    /**
+     * Rolls back the transaction and makes it inactive.
+     *
+     * @return \Interop\Async\Awaitable<\Amp\Postgres\CommandResult>
+     *
+     * @throws \Amp\Postgres\TransactionError
+     */
+    public function rollback(): Awaitable {
+        return new Coroutine($this->doRollback());
+    }
+    
+    public function doRollback(): \Generator {
+        if ($this->executor === null) {
+            throw new TransactionError("The transaction has been committed or rolled back");
+        }
+
+        $executor = $this->executor;
+        $this->executor = null;
+
+        try {
+            $result = yield $executor->query("ROLLBACK");
+        } finally {
+            $this->complete();
+        }
+
+        return $result;
+    }
+
+    /**
+     * Creates a savepoint with the given identifier. WARNING: Identifier is not sanitized, do not pass untrusted data.
+     *
+     * @return \Interop\Async\Awaitable<\Amp\Postgres\CommandResult>
+     *
+     * @throws \Amp\Postgres\TransactionError
+     */
+    public function savepoint(string $identifier): Awaitable {
+        return $this->query("SAVEPOINT " . $identifier);
+    }
+
+    /**
+     * @coroutine
+     *
+     * Rolls back to the savepoint with the given identifier. WARNING: Identifier is not sanitized, do not pass
+     * untrusted data.
+     *
+     * @return \Interop\Async\Awaitable<\Amp\Postgres\CommandResult>
+     *
+     * @throws \Amp\Postgres\TransactionError
+     */
+    public function rollbackTo(string $identifier): Awaitable {
+        return $this->query("ROLLBACK TO " . $identifier);
+    }
+
+    /**
+     * @coroutine
+     *
+     * Releases the savepoint with the given identifier. WARNING: Identifier is not sanitized, do not pass untrusted
+     * data.
+     *
+     * @return \Interop\Async\Awaitable<\Amp\Postgres\CommandResult>
+     *
+     * @throws \Amp\Postgres\TransactionError
+     */
+    public function release(string $identifier): Awaitable {
+        return $this->query("RELEASE SAVEPOINT " . $identifier);
+    }
+}

lib/TransactionError.php

@@ -0,0 +1,5 @@
+<?php declare(strict_types = 1);
+
+namespace Amp\Postgres;
+
+class TransactionError extends \Error {}

lib/TupleResult.php

@@ -0,0 +1,14 @@
+<?php declare(strict_types = 1);
+
+namespace Amp\Postgres;
+
+use Amp\Observer;
+
+abstract class TupleResult extends Observer implements Result {
+    /**
+     * Returns the number of fields (columns) in each row.
+     *
+     * @return int
+     */
+    abstract public function numFields(): int;
+}

lib/functions.php

@@ -0,0 +1,41 @@
+<?php declare(strict_types = 1);
+
+namespace Amp\Postgres;
+
+use Interop\Async\Awaitable;
+
+/**
+ * @param string $connectionString
+ * @param int $timeout
+ *
+ * @return \Interop\Async\Awaitable<\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 {
+    if (\extension_loaded("pq")) {
+        return PqConnection::connect($connectionString, $timeout);
+    }
+    
+    if (\extension_loaded("pgsql")) {
+        return PgSqlConnection::connect($connectionString, $timeout);
+    }
+    
+    throw new \Error("This lib requires either pecl-pq or ext-pgsql");
+}
+
+/**
+ * @param string $connectionString
+ * @param int $maxConnections
+ * @param int $connectTimeout
+ *
+ * @return \Amp\Postgres\Pool
+ */
+function pool(
+    string $connectionString,
+    int $maxConnections = ConnectionPool::DEFAULT_MAX_CONNECTIONS,
+    int $connectTimeout = ConnectionPool::DEFAULT_CONNECT_TIMEOUT
+): Pool {
+    return new ConnectionPool($connectionString, $maxConnections, $connectTimeout);
+}

phpdoc.dist.xml

@@ -0,0 +1,23 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<phpdoc>
+    <title>Amp Postgres</title>
+    <parser>
+        <target>build/docs</target>
+        <encoding>utf8</encoding>
+    </parser>
+    <transformer>
+        <target>build/docs</target>
+    </transformer>
+    <logging>
+        <level>warn</level>
+        <paths>
+            <default>build/log/docs/{DATE}.log</default>
+        </paths>
+    </logging>
+    <transformations>
+        <template name="responsive"/>
+    </transformations>
+    <files>
+        <directory>src</directory>
+    </files>
+</phpdoc>

phpunit.xml.dist

@@ -0,0 +1,29 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<phpunit
+        xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+        xsi:noNamespaceSchemaLocation="http://schema.phpunit.de/4.1/phpunit.xsd"
+        backupGlobals="false"
+        backupStaticAttributes="false"
+        bootstrap="vendor/autoload.php"
+        colors="true"
+        convertErrorsToExceptions="true"
+        convertNoticesToExceptions="true"
+        convertWarningsToExceptions="true"
+        processIsolation="false"
+        stopOnFailure="false"
+>
+    <testsuites>
+        <testsuite name="Icicle Postgres Client">
+            <directory>test</directory>
+        </testsuite>
+    </testsuites>
+    <filter>
+        <whitelist>
+            <directory suffix=".php">src</directory>
+        </whitelist>
+    </filter>
+    <logging>
+        <log type="coverage-html" target="build/coverage" title="Amp" highlight="true"/>
+        <log type="coverage-clover" target="build/logs/clover.xml"/>
+    </logging>
+</phpunit>

test/AbstractConnectionTest.php

@@ -0,0 +1,323 @@
+<?php declare(strict_types = 1);
+
+namespace Amp\Postgres\Test;
+
+use Amp\{ Coroutine, Pause };
+use Amp\Postgres\{ CommandResult, Connection, QueryError, Transaction, TransactionError, TupleResult };
+use Interop\Async\Loop;
+
+abstract class AbstractConnectionTest extends \PHPUnit_Framework_TestCase {
+    /** @var \Amp\Postgres\Connection */
+    protected $connection;
+
+    /**
+     * @return array Start test data for database.
+     */
+    public function getData() {
+        return [
+            ['amphp', 'org'],
+            ['github', 'com'],
+            ['google', 'com'],
+            ['php', 'net'],
+        ];
+    }
+    
+    abstract public function createConnection(string $connectionString): Connection;
+
+    abstract public function getConnectCallable(): callable;
+    
+    public function setUp() {
+        $this->connection = $this->createConnection('host=localhost user=postgres');
+    }
+    
+    public function testQueryWithTupleResult() {
+        \Amp\execute(function () {
+            /** @var \Amp\Postgres\TupleResult $result */
+            $result = yield $this->connection->query("SELECT * FROM test");
+
+            $this->assertInstanceOf(TupleResult::class, $result);
+
+            $this->assertSame(2, $result->numFields());
+            
+            $data = $this->getData();
+
+            for ($i = 0; yield $result->next(); ++$i) {
+                $row = $result->getCurrent();
+                $this->assertSame($data[$i][0], $row['domain']);
+                $this->assertSame($data[$i][1], $row['tld']);
+            }
+        }, Loop::get());
+    }
+
+    public function testQueryWithCommandResult() {
+        \Amp\execute(function () {
+            /** @var \Amp\Postgres\CommandResult $result */
+            $result = yield $this->connection->query("INSERT INTO test VALUES ('canon', 'jp')");
+
+            $this->assertInstanceOf(CommandResult::class, $result);
+            $this->assertSame(1, $result->affectedRows());
+        }, Loop::get());
+    }
+
+    /**
+     * @expectedException \Amp\Postgres\QueryError
+     */
+    public function testQueryWithEmptyQuery() {
+        \Amp\execute(function () {
+            /** @var \Amp\Postgres\CommandResult $result */
+            $result = yield $this->connection->query('');
+        }, Loop::get());
+    }
+
+    /**
+     * @expectedException \Amp\Postgres\QueryError
+     */
+    public function testQueryWithSyntaxError() {
+        \Amp\execute(function () {
+            /** @var \Amp\Postgres\CommandResult $result */
+            $result = yield $this->connection->query("SELECT & FROM test");
+        }, Loop::get());
+    }
+
+    public function testPrepare() {
+        \Amp\execute(function () {
+            $query = "SELECT * FROM test WHERE domain=\$1";
+
+            /** @var \Amp\Postgres\Statement $statement */
+            $statement = yield $this->connection->prepare($query);
+
+            $this->assertSame($query, $statement->getQuery());
+
+            $data = $this->getData()[0];
+
+            /** @var \Amp\Postgres\TupleResult $result */
+            $result = yield $statement->execute($data[0]);
+
+            $this->assertInstanceOf(TupleResult::class, $result);
+
+            $this->assertSame(2, $result->numFields());
+    
+            while (yield $result->next()) {
+                $row = $result->getCurrent();
+                $this->assertSame($data[0], $row['domain']);
+                $this->assertSame($data[1], $row['tld']);
+            }
+        }, Loop::get());
+    }
+
+    public function testExecute() {
+        \Amp\execute(function () {
+            $data = $this->getData()[0];
+
+            /** @var \Amp\Postgres\TupleResult $result */
+            $result = yield $this->connection->execute("SELECT * FROM test WHERE domain=\$1", $data[0]);
+
+            $this->assertInstanceOf(TupleResult::class, $result);
+
+            $this->assertSame(2, $result->numFields());
+
+            while (yield $result->next()) {
+                $row = $result->getCurrent();
+                $this->assertSame($data[0], $row['domain']);
+                $this->assertSame($data[1], $row['tld']);
+            }
+        }, Loop::get());
+    }
+
+    /**
+     * @depends testQueryWithTupleResult
+     */
+    public function testSimultaneousQuery() {
+        $callback = \Amp\coroutine(function ($value) {
+            /** @var \Amp\Postgres\TupleResult $result */
+            $result = yield $this->connection->query("SELECT {$value} as value");
+
+            if ($value) {
+                yield new Pause(100);
+            }
+    
+            while (yield $result->next()) {
+                $row = $result->getCurrent();
+                $this->assertEquals($value, $row['value']);
+            }
+        });
+        
+        \Amp\execute(function () use ($callback) {
+            yield \Amp\all([$callback(0), $callback(1)]);
+        }, Loop::get());
+    }
+
+    /**
+     * @depends testSimultaneousQuery
+     */
+    public function testSimultaneousQueryWithFirstFailing() {
+        $callback = \Amp\coroutine(function ($query) {
+            /** @var \Amp\Postgres\TupleResult $result */
+            $result = yield $this->connection->query($query);
+
+            $data = $this->getData();
+    
+            for ($i = 0; yield $result->next(); ++$i) {
+                $row = $result->getCurrent();
+                $this->assertSame($data[$i][0], $row['domain']);
+                $this->assertSame($data[$i][1], $row['tld']);
+            }
+        });
+    
+        try {
+            \Amp\execute(function () use ($callback) {
+                $failing = $callback("SELECT & FROM test");
+                $successful = $callback("SELECT * FROM test");
+                
+                yield $successful;
+                yield $failing;
+            }, Loop::get());
+        } catch (QueryError $exception) {
+            return;
+        }
+        
+        $this->fail(\sprintf("Test did not throw an instance of %s", QueryError::class));
+    }
+
+    public function testSimultaneousQueryAndPrepare() {
+        $awaitables = [];
+        $awaitables[] = new Coroutine((function () {
+            /** @var \Amp\Postgres\TupleResult $result */
+            $result = yield $this->connection->query("SELECT * FROM test");
+
+            $data = $this->getData();
+    
+            for ($i = 0; yield $result->next(); ++$i) {
+                $row = $result->getCurrent();
+                $this->assertSame($data[$i][0], $row['domain']);
+                $this->assertSame($data[$i][1], $row['tld']);
+            }
+        })());
+
+        $awaitables[] = new Coroutine((function () {
+            /** @var \Amp\Postgres\Statement $statement */
+            $statement = (yield $this->connection->prepare("SELECT * FROM test"));
+
+            /** @var \Amp\Postgres\TupleResult $result */
+            $result = yield $statement->execute();
+
+            $data = $this->getData();
+    
+            for ($i = 0; yield $result->next(); ++$i) {
+                $row = $result->getCurrent();
+                $this->assertSame($data[$i][0], $row['domain']);
+                $this->assertSame($data[$i][1], $row['tld']);
+            }
+        })());
+
+        \Amp\execute(function () use ($awaitables) {
+            yield \Amp\all($awaitables);
+        }, Loop::get());
+    }
+
+    public function testSimultaneousPrepareAndExecute() {
+        $awaitables[] = new Coroutine((function () {
+            /** @var \Amp\Postgres\Statement $statement */
+            $statement = yield $this->connection->prepare("SELECT * FROM test");
+
+            /** @var \Amp\Postgres\TupleResult $result */
+            $result = yield $statement->execute();
+
+            $data = $this->getData();
+    
+            for ($i = 0; yield $result->next(); ++$i) {
+                $row = $result->getCurrent();
+                $this->assertSame($data[$i][0], $row['domain']);
+                $this->assertSame($data[$i][1], $row['tld']);
+            }
+        })());
+
+        $awaitables[] = new Coroutine((function () {
+            /** @var \Amp\Postgres\TupleResult $result */
+            $result = yield $this->connection->execute("SELECT * FROM test");
+
+            $data = $this->getData();
+    
+            for ($i = 0; yield $result->next(); ++$i) {
+                $row = $result->getCurrent();
+                $this->assertSame($data[$i][0], $row['domain']);
+                $this->assertSame($data[$i][1], $row['tld']);
+            }
+        })());
+
+        \Amp\execute(function () use ($awaitables) {
+            yield \Amp\all($awaitables);
+        }, Loop::get());
+    }
+
+    public function testTransaction() {
+        \Amp\execute(function () {
+            $isolation = Transaction::COMMITTED;
+
+            /** @var \Amp\Postgres\Transaction $transaction */
+            $transaction = yield $this->connection->transaction($isolation);
+
+            $this->assertInstanceOf(Transaction::class, $transaction);
+
+            $data = $this->getData()[0];
+
+            $this->assertTrue($transaction->isActive());
+            $this->assertSame($isolation, $transaction->getIsolationLevel());
+
+            yield $transaction->savepoint('test');
+
+            $result = yield $transaction->execute("SELECT * FROM test WHERE domain=\$1 FOR UPDATE", $data[0]);
+
+            yield $transaction->rollbackTo('test');
+
+            yield $transaction->commit();
+
+            $this->assertFalse($transaction->isActive());
+
+            try {
+                $result = yield $transaction->execute("SELECT * FROM test");
+                $this->fail('Query should fail after transaction commit');
+            } catch (TransactionError $exception) {
+                // Exception expected.
+            }
+        }, Loop::get());
+    }
+    
+    public function testConnect() {
+        \Amp\execute(function () {
+            $connect = $this->getConnectCallable();
+            $connection = yield $connect('host=localhost user=postgres');
+            $this->assertInstanceOf(Connection::class, $connection);
+        });
+    }
+    
+    /**
+     * @expectedException \Amp\Postgres\FailureException
+     */
+    public function testConnectInvalidUser() {
+        \Amp\execute(function () {
+            $connect = $this->getConnectCallable();
+            $connection = yield $connect('host=localhost user=invalid', 1);
+        });
+    }
+    
+    /**
+     * @expectedException \Amp\Postgres\FailureException
+     */
+    public function testConnectInvalidConnectionString() {
+        \Amp\execute(function () {
+            $connect = $this->getConnectCallable();
+            $connection = yield $connect('invalid connection string', 1);
+        });
+    }
+    
+    /**
+     * @expectedException \Amp\Postgres\FailureException
+     */
+    public function testConnectInvalidHost() {
+        \Amp\execute(function () {
+            $connect = $this->getConnectCallable();
+            $connection = yield $connect('hostaddr=invalid.host user=postgres', 1);
+        });
+    }
+}

test/AbstractPoolTest.php

@@ -0,0 +1,182 @@
+<?php declare(strict_types = 1);
+
+namespace Amp\Postgres\Test;
+
+use Amp\Postgres\{ CommandResult, Connection, Statement, Transaction, TupleResult };
+use Amp\Success;
+
+abstract class AbstractPoolTest extends \PHPUnit_Framework_TestCase {
+    /**
+     * @param array $connections
+     *
+     * @return \Amp\Postgres\Pool
+     */
+    abstract protected function createPool(array $connections);
+
+    /**
+     * @return \PHPUnit_Framework_MockObject_MockObject|\Amp\Postgres\Connection
+     */
+    private function createConnection() {
+        return $this->createMock(Connection::class);
+    }
+
+    /**
+     * @param int $count
+     *
+     * @return \Amp\Postgres\Connection[]|\PHPUnit_Framework_MockObject_MockObject[]
+     */
+    private function makeConnectionSet($count) {
+        $connections = [];
+
+        for ($i = 0; $i < $count; ++$i) {
+            $connections[] = $this->createConnection();
+        }
+
+        return $connections;
+    }
+
+    /**
+     * @return array
+     */
+    public function getMethodsAndResults() {
+        return [
+            [3, 'query', TupleResult::class, "SELECT * FROM test"],
+            [2, 'query', CommandResult::class, "INSERT INTO test VALUES (1, 7)"],
+            [1, 'prepare', Statement::class, "SELECT * FROM test WHERE id=\$1"],
+            [4, 'execute', TupleResult::class, "SELECT * FROM test WHERE id=\$1 AND time>\$2", 1, time()],
+        ];
+    }
+
+    /**
+     * @dataProvider getMethodsAndResults
+     *
+     * @param int $count
+     * @param string $method
+     * @param string $resultClass
+     * @param mixed ...$params
+     */
+    public function testSingleQuery($count, $method, $resultClass, ...$params) {
+        $result = $this->getMockBuilder($resultClass)
+            ->disableOriginalConstructor()
+            ->getMock();
+
+        $connections = $this->makeConnectionSet($count);
+
+        $connection = $connections[0];
+        $connection->expects($this->once())
+            ->method($method)
+            ->with(...$params)
+            ->will($this->returnValue(new Success($result)));
+
+        $pool = $this->createPool($connections);
+
+        \Amp\execute(function () use ($method, $pool, $params, $result) {
+            $return = yield $pool->{$method}(...$params);
+
+            $this->assertSame($result, $return);
+        });
+    }
+
+    /**
+     * @dataProvider getMethodsAndResults
+     *
+     * @param int $count
+     * @param string $method
+     * @param string $resultClass
+     * @param mixed ...$params
+     */
+    public function testConsecutiveQueries($count, $method, $resultClass, ...$params) {
+        $rounds = 3;
+        $result = $this->getMockBuilder($resultClass)
+            ->disableOriginalConstructor()
+            ->getMock();
+
+        $connections = $this->makeConnectionSet($count);
+
+        foreach ($connections as $connection) {
+            $connection->method($method)
+                ->with(...$params)
+                ->will($this->returnValue(new Success($result)));
+        }
+
+        $pool = $this->createPool($connections);
+
+        
+        \Amp\execute(function () Use ($count, $rounds, $pool, $method, $params) {
+            $awaitables = [];
+    
+            for ($i = 0; $i < $count * $rounds; ++$i) {
+                $awaitables[] = $pool->{$method}(...$params);
+            }
+        });
+    }
+
+    /**
+     * @return array
+     */
+    public function getConnectionCounts() {
+        return array_map(function ($count) { return [$count]; }, range(1, 10));
+    }
+
+    /**
+     * @dataProvider getConnectionCounts
+     *
+     * @param int $count
+     */
+    public function testTransaction($count) {
+        $connections = $this->makeConnectionSet($count);
+
+        $connection = $connections[0];
+        $result = $this->getMockBuilder(Transaction::class)
+            ->disableOriginalConstructor()
+            ->getMock();
+
+        $connection->expects($this->once())
+            ->method('transaction')
+            ->with(Transaction::COMMITTED)
+            ->will($this->returnValue(new Success($result)));
+
+        $pool = $this->createPool($connections);
+        
+        \Amp\execute(function () use ($pool, $result) {
+            $return = yield $pool->transaction(Transaction::COMMITTED);
+            $this->assertInstanceOf(Transaction::class, $return);
+            yield $return->rollback();
+        });
+    }
+
+    /**
+     * @dataProvider getConnectionCounts
+     *
+     * @param int $count
+     */
+    public function testConsecutiveTransactions($count) {
+        $rounds = 3;
+        $result = $this->getMockBuilder(Transaction::class)
+            ->disableOriginalConstructor()
+            ->getMock();
+
+        $connections = $this->makeConnectionSet($count);
+
+        foreach ($connections as $connection) {
+            $connection->method('transaction')
+                ->with(Transaction::COMMITTED)
+                ->will($this->returnCallback(function () use ($result) {
+                    return new Success($result);
+                }));
+        }
+
+        $pool = $this->createPool($connections);
+
+        \Amp\execute(function () use ($count, $rounds, $pool) {
+            $awaitables = [];
+            for ($i = 0; $i < $count * $rounds; ++$i) {
+                $awaitables[] = $pool->transaction(Transaction::COMMITTED);
+            }
+            
+            yield \Amp\all(\Amp\map(function (Transaction $transaction) {
+                return $transaction->rollback();
+            }, $awaitables));
+        });
+    }
+}

test/AggregatePoolTest.php

@@ -0,0 +1,31 @@
+<?php declare(strict_types = 1);
+
+namespace Amp\Postgres\Test;
+
+use Amp\Postgres\AggregatePool;
+
+class AggregatePoolTest extends AbstractPoolTest
+{
+    /**
+     * @param array $connections
+     *
+     * @return \Amp\Postgres\Pool
+     */
+    protected function createPool(array $connections) {
+        $mock = $this->getMockBuilder(AggregatePool::class)
+            ->setConstructorArgs(['', 0, count($connections)])
+            ->setMethods(['createConnection'])
+            ->getMock();
+
+        $mock->method('createConnection')
+            ->will($this->returnCallback(function () {
+                $this->fail('The createConnection() method should not be called.');
+            }));
+
+        foreach ($connections as $connection) {
+            $mock->addConnection($connection);
+        }
+
+        return $mock;
+    }
+}

test/ConnectionPoolTest.php

@@ -0,0 +1,30 @@
+<?php declare(strict_types = 1);
+
+namespace Amp\Postgres\Test;
+
+use Amp\Postgres\ConnectionPool;
+use Amp\Success;
+use Interop\Async\Awaitable;
+
+class ConnectionPoolTest extends AbstractPoolTest
+{
+    /**
+     * @param array $connections
+     *
+     * @return \Amp\Postgres\Pool
+     */
+    protected function createPool(array $connections) {
+        $mock = $this->getMockBuilder(ConnectionPool::class)
+            ->setConstructorArgs(['connection string', \count($connections)])
+            ->setMethods(['createConnection'])
+            ->getMock();
+
+        $mock->method('createConnection')
+            ->will($this->returnCallback(function () use ($connections): Awaitable {
+                static $count = 0;
+                return new Success($connections[$count++]);
+            }));
+
+        return $mock;
+    }
+}

test/FunctionsTest.php

@@ -0,0 +1,42 @@
+<?php declare(strict_types = 1);
+
+namespace Amp\Postgres\Test;
+
+use Amp\Postgres\{ Connection, function connect };
+
+class FunctionsTest extends \PHPUnit_Framework_TestCase
+{
+    public function testConnect() {
+        \Amp\execute(function () {
+            $connection = yield connect('host=localhost user=postgres', 1);
+            $this->assertInstanceOf(Connection::class, $connection);
+        });
+    }
+
+    /**
+     * @expectedException \Amp\Postgres\FailureException
+     */
+    public function testConnectInvalidUser() {
+        \Amp\execute(function () {
+            $connection = yield connect('host=localhost user=invalid', 1);
+        });
+    }
+
+    /**
+     * @expectedException \Amp\Postgres\FailureException
+     */
+    public function testConnectInvalidConnectionString() {
+        \Amp\execute(function () {
+            $connection = yield connect('invalid connection string', 1);
+        });
+    }
+
+    /**
+     * @expectedException \Amp\Postgres\FailureException
+     */
+    public function testConnectInvalidHost() {
+        \Amp\execute(function () {
+            $connection = yield connect('hostaddr=invalid.host user=postgres', 1);
+        });
+    }
+}

test/PgSqlConnectionTest.php

@@ -0,0 +1,40 @@
+<?php declare(strict_types = 1);
+
+namespace Amp\Postgres\Test;
+
+use Amp\Postgres\{ Connection, PgSqlConnection };
+
+class PgSqlConnectionTest extends AbstractConnectionTest {
+    /** @var resource PostgreSQL connection resource. */
+    protected $handle;
+
+    public function createConnection(string $connectionString): Connection {
+        $this->handle = \pg_connect($connectionString);
+        $socket = \pg_socket($this->handle);
+    
+        $result = \pg_query($this->handle, "CREATE TABLE test (domain VARCHAR(63), tld VARCHAR(63), PRIMARY KEY (domain, tld))");
+    
+        if (!$result) {
+            $this->fail('Could not create test table.');
+        }
+    
+        foreach ($this->getData() as $row) {
+            $result = \pg_query_params($this->handle, "INSERT INTO test VALUES (\$1, \$2)", $row);
+        
+            if (!$result) {
+                $this->fail('Could not insert test data.');
+            }
+        }
+    
+        return new PgSqlConnection($this->handle, $socket);
+    }
+    
+    public function getConnectCallable(): callable {
+        return [PgSqlConnection::class, 'connect'];
+    }
+    
+    public function tearDown() {
+        \pg_query($this->handle, "ROLLBACK");
+        \pg_query($this->handle, "DROP TABLE test");
+    }
+}

test/PqConnectionTest.php

@@ -0,0 +1,39 @@
+<?php declare(strict_types = 1);
+
+namespace Amp\Postgres\Test;
+
+use Amp\Postgres\{ Connection, PqConnection };
+
+class PqConnectionTest extends AbstractConnectionTest {
+    /** @var resource PostgreSQL connection resource. */
+    protected $handle;
+
+    public function createConnection(string $connectionString): Connection {
+        $this->handle = new \pq\Connection($connectionString);
+    
+        $result = $this->handle->exec("CREATE TABLE test (domain VARCHAR(63), tld VARCHAR(63), PRIMARY KEY (domain, tld))");
+    
+        if (!$result) {
+            $this->fail('Could not create test table.');
+        }
+    
+        foreach ($this->getData() as $row) {
+            $result = $this->handle->execParams("INSERT INTO test VALUES (\$1, \$2)", $row);
+        
+            if (!$result) {
+                $this->fail('Could not insert test data.');
+            }
+        }
+    
+        return new PqConnection($this->handle);
+    }
+    
+    public function getConnectCallable(): callable {
+        return [PqConnection::class, 'connect'];
+    }
+    
+    public function tearDown() {
+        $this->handle->exec("ROLLBACK");
+        $this->handle->exec("DROP TABLE test");
+    }
+}