PgAsync

Asynchronous Reactive Postgres Library for PHP (Non-blocking)

What it is

This is an asynchronous Postgres library for PHP. Observables are returned by the query methods allowing asynchronous row-by-row data handling (and other Rx operators on the data) See Rx.PHP. Network and event processing is handled by ReactPHP.

This is a pure PHP implementation (you don't need Postgres extensions to use it).

Example - Simple Query

$client = new PgAsync\Client([
    "host" => "127.0.0.1",
    "port" => "5432",
    "user"     => "matt",
    "database" => "matt"
]);

$client->query('SELECT * FROM channel')->subscribe(new \Rx\Observer\CallbackObserver(
    function ($row) {
        var_dump($row);
    },
    function ($e) {
        echo "Failed.\n";
    },
    function () {
        echo "Complete.\n";
    }
));

Example - parameterized query

$client = new PgAsync\Client([
     "host" => "127.0.0.1",
     "port" => "5432",
     "user"     => "matt",
     "database" => "matt",
     "auto_disconnect" => true //This option will force the client to disconnect as soon as it completes.  The connection will not be returned to the connection pool.

]);

$client->executeStatement('SELECT * FROM channel WHERE id = $1', ['5'])
    ->subscribe(new \Rx\Observer\CallbackObserver(
        function ($row) {
            var_dump($row);
        },
        function ($e) {
            echo "Failed.\n";
        },
        function () {
            echo "Complete.\n";
        }
    ));

Install

With composer install into you project with:

Install pgasync: composer require voryx/pgasync:dev-master

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

What's next

• Add more testing
• Transactions
• Take over the world

Keep in mind

This is an asynchronous library. If you begin 3 queries (subscribe to their observable):

$client->query("SELECT * FROM table1")->subscribe(...);
$client->query("SELECT * FROM table2")->subscribe(...);
$client->query("SELECT * FROM table3")->subscribe(...);

It will start all of them almost simultaneously (and you will begin receiving rows on all 3 before any of them have completed). This can be great if you want to run 3 queries at the same time, but if you have some queries that need information that was modified by other statements, this can cause a race condition:

$client->query("INSERT INTO invoices(inv_no, customer_id, amount) VALUES('1234A', 1, 35.75)")->subscribe(...);
$client->query("SELECT SUM(amount) AS balance FROM invoices WHERE customer_id = 1")->subscribe(...);

In the above situation, your balance may or may not include the invoice inserted on the first line.

You can avoid this by using the Rx concat* operator to only start up the second observable after the first has completed:

$insert = $client->query("INSERT INTO invoices(inv_no, customer_id, amount) VALUES('1234A', 1, 35.75)");
$select = $client->query("SELECT SUM(amount) AS balance FROM invoices WHERE customer_id = 1");

$insert
    ->concat($select)
    ->subscribe(...);