backmerge

.github/workflows/ci.yml

@@ -0,0 +1,63 @@
+name: Continuous Integration
+on:
+  push:
+    branches:
+      - 'master'
+      - 'refs/heads/[0-9]+.[0-9]+.[0-9]+'
+  pull_request:
+jobs:
+  supported-versions-matrix:
+    name: Supported Versions Matrix
+    runs-on: ubuntu-latest
+    outputs:
+      version: ${{ steps.supported-versions-matrix.outputs.version }}
+    steps:
+      - uses: actions/checkout@v1
+      - id: supported-versions-matrix
+        uses: WyriHaximus/github-action-composer-php-versions-in-range@v1
+  tests:
+    services:
+      postgres:
+        image: postgres:${{ matrix.postgres }}
+        env:
+          POSTGRES_PASSWORD: postgres
+          POSTGRES_INITDB_ARGS: --auth-host=md5
+        # Set health checks to wait until postgres has started
+        options: >-
+          --health-cmd pg_isready
+          --health-interval 10s
+          --health-timeout 5s
+          --health-retries 5
+        ports:
+          - 5432:5432
+    name: Testing on PHP ${{ matrix.php }} with ${{ matrix.composer }} dependency preference against Postgres ${{ matrix.postgres }}
+    strategy:
+      fail-fast: false
+      matrix:
+        php: ${{ fromJson(needs.supported-versions-matrix.outputs.version) }}
+        postgres: [12, 13, 14, 15]
+        composer: [lowest, locked, highest]
+    needs:
+      - supported-versions-matrix
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v3
+      - run: |
+          PGPASSWORD=postgres psql -h localhost -U postgres -c "CREATE USER pgasync"
+          PGPASSWORD=postgres psql -h localhost -U postgres -c "ALTER ROLE pgasync PASSWORD 'pgasync'"
+          PGPASSWORD=postgres psql -h localhost -U postgres -c "CREATE USER pgasyncpw"
+          PGPASSWORD=postgres psql -h localhost -U postgres -c "ALTER ROLE pgasyncpw PASSWORD 'example_password'"
+          PGPASSWORD=postgres psql -h localhost -U postgres -c "CREATE USER scram_user"
+          PGPASSWORD=postgres psql -h localhost -U postgres -c "SET password_encryption='scram-sha-256';ALTER ROLE scram_user PASSWORD 'scram_password'"
+          PGPASSWORD=postgres psql -h localhost -U postgres -c "CREATE DATABASE pgasync_test OWNER pgasync"
+          PGPASSWORD=pgasync psql -h localhost -U pgasync -f tests/test_db.sql pgasync_test
+#          PGPASSWORD=postgres cat tests/test_db.sql | xargs -I % psql -h localhost -U postgres -c "%"
+      - uses: shivammathur/setup-php@v2
+        with:
+          php-version: ${{ matrix.php }}
+          coverage: xdebug
+      - uses: ramsey/composer-install@v2
+        with:
+          dependency-versions: ${{ matrix.composer }}
+#      - run: vendor/bin/phpunit --testdox
+      - run: vendor/bin/phpunit

.gitignore

@@ -1,2 +1,3 @@
 /composer.lock
 /vendor
+/docker/database

.travis.yml

@@ -2,14 +2,19 @@ language: php
 sudo: required
 
 php:
-  - 7.0
-  - 7.1
+  - 7.2
+  - 7.3
+  - 7.4
 
-addons:
-  postgresql: "9.3"
+services:
+  - docker
 
 install:
   - composer install
 
+before_script:
+  - docker-compose -f docker/docker-compose.yml up -d
+  - sh docker/waitForPostgres.sh
+
 script:
-  - vendor/bin/phpunit
+  - vendor/bin/phpunit --testdox

README.md

@@ -20,7 +20,7 @@ $client = new PgAsync\Client([
     "database" => "matt"
 ]);
 
-$client->query('SELECT * FROM channel')->subscribe(new \Rx\Observer\CallbackObserver(
+$client->query('SELECT * FROM channel')->subscribe(
     function ($row) {
         var_dump($row);
     },
@@ -30,7 +30,7 @@ $client->query('SELECT * FROM channel')->subscribe(new \Rx\Observer\CallbackObse
     function () {
         echo "Complete.\n";
     }
-));
+);
 
 
 ```
@@ -48,7 +48,7 @@ $client = new PgAsync\Client([
 ]);
 
 $client->executeStatement('SELECT * FROM channel WHERE id = $1', ['5'])
-    ->subscribe(new \Rx\Observer\CallbackObserver(
+    ->subscribe(
         function ($row) {
             var_dump($row);
         },
@@ -58,15 +58,32 @@ $client->executeStatement('SELECT * FROM channel WHERE id = $1', ['5'])
         function () {
             echo "Complete.\n";
         }
-    ));
+    );
 
 ```
 
+## Example - LISTEN/NOTIFY
+```php
+$client = new PgAsync\Client([
+     "host" => "127.0.0.1",
+     "port" => "5432",
+     "user"     => "matt",
+     "database" => "matt"
+]);
+
+$client->listen('some_channel')
+    ->subscribe(function (\PgAsync\Message\NotificationResponse $message) {
+        echo $message->getChannelName() . ': ' . $message->getPayload() . "\n";
+    });
+
+$client->query("NOTIFY some_channel, 'Hello World'")->subscribe();
+```
+
 ## Install
 With [composer](https://getcomposer.org/) install into you project with:
 
 Install pgasync:
-```composer require voryx/pgasync:dev-master```
+```composer require voryx/pgasync```
 
 ## What it can do
 - Run queries (CREATE, UPDATE, INSERT, SELECT, DELETE)
@@ -76,7 +93,7 @@ Install pgasync:
 - Connection pooling (basic pooling)
 
 ## What it can't quite do yet
-- Transactions
+- Transactions (Actually though, just grab a connection and you can run your transaction on that single connection)
 
 ## What's next
 - Add more testing
@@ -112,3 +129,17 @@ $insert
     ->concat($select)
     ->subscribe(...);
 ```
+
+## Testing
+
+We use docker to run a postgresql instance for testing. To run locally,
+just install docker and run the following command from the project root:
+```bash
+docker-compose -f docker/docker-compose.yml up -d
+```
+If you need to reset the database, just stop the docker instance and delete
+the `docker/database` directory. Restart the docker with the above command and it will
+initialize the database again.
+
+The tests do not change the ending structure of the database, so you should not
+normally need to do this.

composer.json

@@ -33,12 +33,18 @@
     }
   },
   "require": {
-    "voryx/event-loop": "^2.0.2",
+    "php": ">=7.0.0",
+    "voryx/event-loop": "^3.0 || ^2.0.2",
     "reactivex/rxphp": "^2.0",
     "react/socket": "^1.0 || ^0.8 || ^0.7",
     "evenement/evenement": "^2.0 | ^3.0"
   },
   "require-dev": {
-    "phpunit/phpunit": "^5.7"
+    "phpunit/phpunit": ">=8.5.23 || ^6.5.5",
+    "react/dns": "^1.0"
+  },
+  "scripts": {
+    "docker-up": "cd docker && docker-compose up -d",
+    "docker-down": "cd docker && docker-compose down"
   }
 }

docker/docker-compose.yml

@@ -0,0 +1,23 @@
+version: '3.7'
+
+services:
+  pgasync-postgres:
+    container_name: pgasync-postgres
+    image: postgres:11
+    environment:
+      - PGDATA=/database
+      - POSTGRES_PASSWORD=some_password
+      - POSTGRES_INITDB_ARGS=--auth-host=md5
+      - TZ=America/New_York
+    volumes:
+      - .:/app
+      - ./database:/database
+      - ./../tests/test_db.sql:/app/test_db.sql
+      - ./docker-entrypoint-initdb.d:/docker-entrypoint-initdb.d
+    ports:
+      - "5432:5432"
+
+configs:
+  pg_hba:
+    file: pg_hba_new.conf
+

docker/docker-entrypoint-initdb.d/init.sh

@@ -0,0 +1,19 @@
+#!/bin/bash
+set -x
+
+echo "Running as $USER in $PWD"
+
+createuser -U postgres --createdb pgasync
+createuser -U postgres --createdb pgasyncpw
+createuser -U postgres --createdb scram_user
+psql -U postgres -c "ALTER ROLE pgasyncpw PASSWORD 'example_password'"
+psql -U postgres -c "SET password_encryption='scram-sha-256'; ALTER ROLE scram_user PASSWORD 'scram_password'"
+
+cd /app
+cp pg_hba_new.conf database/pg_hba.conf
+
+createdb -U pgasync pgasync_test
+
+psql -U pgasync -f test_db.sql pgasync_test
+
+

docker/pg_hba_new.conf

@@ -0,0 +1,16 @@
+# TYPE  DATABASE        USER            ADDRESS                 METHOD
+
+# "local" is for Unix domain socket connections only
+local   all             all                                     trust
+# IPv4 local connections:
+host    all             all             127.0.0.1/32            trust
+# IPv6 local connections:
+host    all             all             ::1/128                 trust
+# Allow replication connections from localhost, by a user with the
+# replication privilege.
+local   replication     all                                     trust
+host    replication     all             127.0.0.1/32            trust
+host    replication     all             ::1/128                 trust
+
+host all pgasync all trust
+host all all all md5

example/ListenNotify.php

@@ -0,0 +1,20 @@
+<?php
+
+require_once __DIR__ . '/bootstrap.php';
+
+$client = new PgAsync\Client([
+    'host'     => '127.0.0.1',
+    'port'     => '5432',
+    'user'     => 'matt',
+    'database' => 'matt',
+]);
+
+$client->listen('some_channel')
+    ->subscribe(function (\PgAsync\Message\NotificationResponse $message) {
+        echo $message->getChannelName() . ': ' . $message->getPayload() . "\n";
+    });
+
+\Rx\Observable::timer(1000)
+    ->flatMapTo($client->query("NOTIFY some_channel, 'Hello World'"))
+    ->subscribe();
+

phpunit.xml

@@ -1,5 +1,6 @@
 <?xml version="1.0" encoding="utf-8" ?>
-<phpunit
+<phpunit xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+         xsi:noNamespaceSchemaLocation="./vendor/phpunit/phpunit/phpunit.xsd"
         backupGlobals="false"
         backupStaticAttributes="false"
         colors="true"
@@ -8,7 +9,6 @@
         convertWarningsToExceptions="true"
         processIsolation="false"
         stopOnFailure="false"
-        syntaxCheck="false"
         bootstrap="tests/bootstrap.php">
     <testsuites>
         <testsuite name="PgAsync Tests">

src/PgAsync/Client.php

@@ -2,9 +2,11 @@
 
 namespace PgAsync;
 
+use PgAsync\Message\NotificationResponse;
 use React\EventLoop\LoopInterface;
 use React\Socket\ConnectorInterface;
 use Rx\Observable;
+use Rx\Subject\Subject;
 
 class Client
 {
@@ -30,10 +32,16 @@ class Client
     /** @var int */
     private $maxConnections = 5;
 
+    /** @var Subject[] */
+    private $listeners = [];
+
+    /** @var Connection */
+    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();
+        $this->connector = $connector;
 
         if (isset($parameters['auto_disconnect'])) {
             $this->autoDisconnect = $parameters['auto_disconnect'];
@@ -71,7 +79,7 @@ public function executeStatement(string $queryString, array $parameters = [])
         });
     }
 
-    private function getLeastBusyConnection() : Connection
+    private function getLeastBusyConnection(): Connection
     {
         if (count($this->connections) === 0) {
             // try to spin up another connection to return
@@ -133,9 +141,9 @@ private function createNewConnection()
         $this->connections[] = $connection;
 
         $connection->on('close', function () use ($connection) {
-            $this->connections = array_filter($this->connections, function ($c) use ($connection) {
+            $this->connections = array_values(array_filter($this->connections, function ($c) use ($connection) {
                 return $connection !== $c;
-            });
+            }));
         });
 
         return $connection;
@@ -158,4 +166,42 @@ public function closeNow()
             $connection->disconnect();
         }
     }
+
+    public function listen(string $channel): Observable
+    {
+        if (isset($this->listeners[$channel])) {
+            return $this->listeners[$channel];
+        }
+
+        $unlisten = function () use ($channel) {
+            $this->listenConnection->query('UNLISTEN ' . $channel)->subscribe();
+
+            unset($this->listeners[$channel]);
+
+            if (empty($this->listeners)) {
+                $this->listenConnection->disconnect();
+                $this->listenConnection = null;
+            }
+        };
+
+        $this->listeners[$channel] = Observable::defer(function () use ($channel) {
+            if ($this->listenConnection === null) {
+                $this->listenConnection = $this->createNewConnection();
+            }
+
+            if ($this->listenConnection === null) {
+                throw new \Exception('Could not get new connection to listen on.');
+            }
+
+            return $this->listenConnection->query('LISTEN ' . $channel)
+                ->merge($this->listenConnection->notifications())
+                ->filter(function (NotificationResponse $message) use ($channel) {
+                    return $message->getChannelName() === $channel;
+                });
+        })
+            ->finally($unlisten)
+            ->share();
+
+        return $this->listeners[$channel];
+    }
 }