[WIP] Add transaction support

README.md

@@ -114,19 +114,72 @@ With [composer](https://getcomposer.org/) install into you project with:
 Install pgasync:
 ```composer require voryx/pgasync```
 
+## Example - Transactions
+
+When using transactions with connection pooling, configure `max_connections` so normal
+queries and transactions share the pool. Transactions reserve a single connection for
+their duration; if the pool is full, both transactions and regular queries wait until a
+connection becomes available.
+
+```php
+$client = new PgAsync\Client([
+    "host"            => "127.0.0.1",
+    "port"            => "5432",
+    "user"            => "matt",
+    "database"        => "matt",
+    "max_connections" => 5,
+]);
+```
+
+Use `transaction()` for a scoped transaction that auto-commits on success and rolls back on failure:
+
+```php
+$client->transaction(function (PgAsync\Transaction $tx) {
+    return $tx->executeStatement(
+        'INSERT INTO invoices(inv_no, customer_id, amount) VALUES ($1, $2, $3)',
+        ['1234A', 1, 35.75]
+    )->concat(
+        $tx->query('SELECT SUM(amount) AS balance FROM invoices WHERE customer_id = 1')
+    );
+})->subscribe(
+    function () {
+        echo "Committed.\n";
+    },
+    function ($e) {
+        echo "Rolled back: " . $e->getMessage() . "\n";
+    }
+);
+```
+
+For manual control, use `beginTransaction()` and call `commit()` or `rollback()` yourself:
+
+```php
+$client->beginTransaction()->flatMap(function (PgAsync\Transaction $tx) {
+    return $tx->executeStatement('INSERT INTO channel(name) VALUES ($1)', ['example'])
+        ->concat($tx->query('SELECT COUNT(*) AS c FROM channel'))
+        ->concat($tx->commit());
+})->subscribe(
+    function () {
+        echo "Committed.\n";
+    },
+    function ($e) {
+        echo "Failed: " . $e->getMessage() . "\n";
+    }
+);
+```
+
+All queries inside a transaction run on a single reserved connection. Regular `$client->query()` calls may use other pooled connections and run in parallel. When every connection is reserved for an open transaction, further queries and transactions wait until one is released.
+
 ## What it can do
 - Run queries (CREATE, UPDATE, INSERT, SELECT, DELETE)
 - Queue commands
 - Return results asynchronously (using Observables - you get data one row at a time as it comes from the db server)
 - Prepared statements (as parameterized queries)
 - Connection pooling (basic pooling)
-
-## What it can't quite do yet
-- Transactions (Actually though, just grab a connection and you can run your transaction on that single connection)
+- Transactions (scoped and explicit)
 
 ## What's next
 - Add more testing
-- Transactions
 - Take over the world
 
 ## Keep in mind

composer.json

@@ -42,6 +42,7 @@
     "wyrihaximus/react-opportunistic-tls": "^1.0.0"
   },
   "require-dev": {
+    "phpstan/phpstan": "^1.12",
     "phpunit/phpunit": ">=8.5.23 || ^6.5.5",
     "react/dns": "^1.12.0"
   },

example/Transaction.php

@@ -0,0 +1,25 @@
+<?php
+require_once __DIR__ . '/bootstrap.php';
+
+$client = new PgAsync\Client([
+    "host"     => "127.0.0.1",
+    "port"     => "5432",
+    "user"     => "matt",
+    "database" => "matt"
+]);
+
+$client->transaction(function (PgAsync\Transaction $tx) {
+    return $tx->executeStatement(
+        'INSERT INTO channel(name, description) VALUES ($1, $2)',
+        ['Test Name', 'Inserted inside a scoped transaction']
+    )->concat(
+        $tx->query("SELECT name FROM channel WHERE name = 'Test Name'")
+    );
+})->subscribe(
+    function () {
+        echo "Transaction committed.\n";
+    },
+    function ($e) {
+        echo "Transaction failed: " . $e->getMessage() . "\n";
+    }
+);

src/PgAsync/Client.php

@@ -4,154 +4,143 @@
 
 use PgAsync\Message\NotificationResponse;
 use React\EventLoop\LoopInterface;
+use React\Promise\PromiseInterface;
 use React\Socket\ConnectorInterface;
 use Rx\Observable;
 use Rx\Subject\Subject;
 
 class Client
 {
-    /** @var  string */
-    protected $connectString;
-
     /** @var LoopInterface */
     protected $loop;
 
-    protected $params = [];
-
-    private $parameters = [];
-
-    /** @var Connection[] */
-    private $connections = [];
-
-    /** @var boolean */
-    private $autoDisconnect;
-
-    /** @var ConnectorInterface */
-    private $connector;
-
-    /** @var int */
-    private $maxConnections = 5;
+    /** @var ConnectionPool */
+    private $connectionPool;
 
     /** @var Subject[] */
     private $listeners = [];
 
-    /** @var Connection */
+    /** @var Connection|null */
     private $listenConnection;
 
     public function __construct(array $parameters, ?LoopInterface $loop = null, ?ConnectorInterface $connector = null)
     {
-        $this->loop      = $loop ?: \EventLoop\getLoop();
-        $this->connector = $connector;
+        $this->loop = $loop ?: \EventLoop\getLoop();
+
+        $autoDisconnect = false;
+        $maxConnections = 5;
 
         if (isset($parameters['auto_disconnect'])) {
-            $this->autoDisconnect = $parameters['auto_disconnect'];
+            $autoDisconnect = $parameters['auto_disconnect'];
         }
 
         if (isset($parameters['max_connections'])) {
             if (!is_int($parameters['max_connections'])) {
                 throw new \InvalidArgumentException('`max_connections` must an be integer greater than zero.');
             }
-            $this->maxConnections = $parameters['max_connections'];
+            $maxConnections = $parameters['max_connections'];
             unset($parameters['max_connections']);
-            if ($this->maxConnections < 1) {
+            if ($maxConnections < 1) {
                 throw new \InvalidArgumentException('`max_connections` must be greater than zero.');
             }
         }
 
-        $this->parameters = $parameters;
+        $this->connectionPool = new ConnectionPool(
+            $parameters,
+            $this->loop,
+            $connector,
+            $autoDisconnect,
+            $maxConnections
+        );
     }
 
     public function query($s)
     {
-        return Observable::defer(function () use ($s) {
-            $conn = $this->getLeastBusyConnection();
-
+        return $this->connectionPool->acquire(false)->flatMap(function (Connection $conn) use ($s) {
             return $conn->query($s);
         });
     }
 
-    public function executeStatement(string $queryString, array $parameters = [])
+    public function beginTransaction(): Observable
     {
-        return Observable::defer(function () use ($queryString, $parameters) {
-            $conn = $this->getLeastBusyConnection();
+        return $this->connectionPool->acquire(true)->flatMap(function (Connection $connection) {
+            return $connection->queryUntilReady('BEGIN')->flatMap(function ($status) use ($connection) {
+                if ($status !== 'T') {
+                    $this->connectionPool->release($connection);
 
-            return $conn->executeStatement($queryString, $parameters);
+                    return Observable::error(new \RuntimeException('Expected transaction status T after BEGIN'));
+                }
+
+                return Observable::of($this->createTransaction($connection));
+            })->catch(function (\Throwable $e) use ($connection) {
+                $this->connectionPool->release($connection);
+
+                return Observable::error($e);
+            });
         });
     }
 
-    private function getLeastBusyConnection(): Connection
+    private function createTransaction(Connection $connection): Transaction
     {
-        if (count($this->connections) === 0) {
-            // try to spin up another connection to return
-            $conn = $this->createNewConnection();
-            if ($conn === null) {
-                throw new \Exception('There are no connections. Cannot find least busy one and could not create a new one.');
-            }
-
-            return $conn;
-        }
-
-        $min = $this->connections[0];
-
-        foreach ($this->connections as $connection) {
-            // if this connection is idle - just return it
-            if ($connection->getBacklogLength() === 0 && $connection->getState() === Connection::STATE_READY) {
-                return $connection;
+        return new Transaction(
+            $connection,
+            function () use ($connection) {
+                $this->connectionPool->release($connection);
             }
+        );
+    }
 
-            if ($min->getBacklogLength() > $connection->getBacklogLength()) {
-                $min = $connection;
+    /**
+     * @param callable(Transaction): (Observable|PromiseInterface|mixed) $fn
+     */
+    public function transaction(callable $fn): Observable
+    {
+        return $this->beginTransaction()->flatMap(function (Transaction $tx) use ($fn) {
+            try {
+                $result = $fn($tx);
+            } catch (\Throwable $e) {
+                return $tx->rollback()->concat(Observable::error($e));
             }
-        }
 
-        if (count($this->connections) < $this->maxConnections) {
-            return $this->createNewConnection();
-        }
+            return $this->normalizeTransactionResult($result)
+                ->concat($tx->commit())
+                ->catch(function (\Throwable $e) use ($tx) {
+                    return $tx->rollback()->concat(Observable::error($e));
+                });
+        });
+    }
 
-        return $min;
+    public function executeStatement(string $queryString, array $parameters = [])
+    {
+        return $this->connectionPool->acquire(false)->flatMap(function (Connection $conn) use ($queryString, $parameters) {
+            return $conn->executeStatement($queryString, $parameters);
+        });
     }
 
-    public function getIdleConnection(): Connection
+    private function normalizeTransactionResult($result): Observable
     {
-        // we want to get the first available one
-        // this will keep the connections at the front the busiest
-        // and then we can add an idle timer to the connections
-        foreach ($this->connections as $connection) {
-            // need to figure out different states (in trans etc.)
-            if ($connection->getState() === Connection::STATE_READY) {
-                return $connection;
-            }
+        if ($result instanceof Observable) {
+            return $result;
         }
 
-        if (count($this->connections) >= $this->maxConnections) {
-            return null;
+        if ($result instanceof PromiseInterface) {
+            return Observable::fromPromise($result);
         }
 
-        return $this->createNewConnection();
+        return Observable::of($result);
     }
 
-    private function createNewConnection()
+    /**
+     * @return Connection|null
+     */
+    public function getIdleConnection()
     {
-        // no idle connections were found - spin up new one
-        $connection = new Connection($this->parameters, $this->loop, $this->connector);
-        if ($this->autoDisconnect) {
-            return $connection;
-        }
-
-        $this->connections[] = $connection;
-
-        $connection->on('close', function () use ($connection) {
-            $this->connections = array_values(array_filter($this->connections, function ($c) use ($connection) {
-                return $connection !== $c;
-            }));
-        });
-
-        return $connection;
+        return $this->connectionPool->getIdleConnection();
     }
 
     public function getConnectionCount(): int
     {
-        return count($this->connections);
+        return $this->connectionPool->getConnectionCount();
     }
 
     /**
@@ -162,9 +151,7 @@ public function getConnectionCount(): int
      */
     public function closeNow()
     {
-        foreach ($this->connections as $connection) {
-            $connection->disconnect();
-        }
+        $this->connectionPool->closeAll();
     }
 
     public function listen(string $channel): Observable
@@ -186,7 +173,7 @@ public function listen(string $channel): Observable
 
         $this->listeners[$channel] = Observable::defer(function () use ($channel) {
             if ($this->listenConnection === null) {
-                $this->listenConnection = $this->createNewConnection();
+                $this->listenConnection = $this->connectionPool->createConnection();
             }
 
             if ($this->listenConnection === null) {