Minor executor changes and test updates

Author: trowski

lib/PgSqlExecutor.php

@@ -50,13 +50,16 @@ public function __construct($handle, $socket) {
 
             while ($result = \pg_get_notify($handle, \PGSQL_ASSOC)) {
                 $channel = $result["message"];
-                if (isset($listeners[$channel])) {
-                    $notification = new Notification;
-                    $notification->channel = $channel;
-                    $notification->pid = $result["pid"];
-                    $notification->payload = $result["payload"];
-                    $listeners[$channel]->emit($notification);
+
+                if (!isset($listeners[$channel])) {
+                    return;
                 }
+
+                $notification = new Notification;
+                $notification->channel = $channel;
+                $notification->pid = $result["pid"];
+                $notification->payload = $result["payload"];
+                $listeners[$channel]->emit($notification);
             }
 
             if ($deferred === null) {
@@ -76,7 +79,7 @@ public function __construct($handle, $socket) {
 
         $this->await = Loop::onWritable($socket, static function ($watcher) use (&$deferred, $handle) {
             $flush = \pg_flush($handle);
-            if (0 === $flush) {
+            if ($flush === 0) {
                 return; // Not finished sending data, listen again.
             }
 
@@ -231,7 +234,7 @@ public function listen(string $channel): Promise {
                 throw new QueryError(\sprintf("Already listening on channel '%s'", $channel));
             }
 
-            $this->listeners[$channel] = $emitter =  new Emitter;
+            $this->listeners[$channel] = $emitter = new Emitter;
 
             try {
                 yield $this->query(\sprintf("LISTEN %s", $channel));
@@ -265,9 +268,7 @@ private function unlisten(string $channel): Promise {
         }
 
         $promise = $this->query(\sprintf("UNLISTEN %s", $channel));
-        $promise->onResolve(function () use ($emitter) {
-            $emitter->complete();
-        });
+        $promise->onResolve([$emitter, "complete"]);
         return $promise;
     }
 }

lib/PqExecutor.php

@@ -284,9 +284,7 @@ private function unlisten(string $channel): Promise {
         }
 
         $promise = new Coroutine($this->send([$this->handle, "unlistenAsync"], $channel));
-        $promise->onResolve(function () use ($emitter) {
-            $emitter->complete();
-        });
+        $promise->onResolve([$emitter, "complete"]);
         return $promise;
     }
 }

test/AbstractConnectionTest.php

@@ -10,6 +10,7 @@
 use Amp\Postgres\TransactionError;
 use Amp\Postgres\TupleResult;
 use PHPUnit\Framework\TestCase;
+use function Amp\asyncCall;
 
 abstract class AbstractConnectionTest extends TestCase {
     /** @var \Amp\Postgres\Connection */
@@ -191,18 +192,21 @@ public function testListen() {
 
             $this->assertInstanceOf(Listener::class, $listener);
 
-            yield $this->connection->query(\sprintf("NOTIFY %s, '%s'", $channel, '0'));
-            yield $this->connection->query(\sprintf("NOTIFY %s, '%s'", $channel, '1'));
+            asyncCall(function () use ($channel) {
+                yield $this->connection->query(\sprintf("NOTIFY %s, '%s'", $channel, '0'));
+                yield $this->connection->query(\sprintf("NOTIFY %s, '%s'", $channel, '1'));
+            });
 
             $count = 0;
-            Loop::defer(function () use (&$count, $listener) {
+            Loop::delay(100, function () use ($listener) {
                 $listener->unlisten();
-                $this->assertSame(2, $count);
             });
 
             while (yield $listener->advance()) {
                 $this->assertSame($listener->getCurrent()->payload, (string) $count++);
             }
+
+            $this->assertSame(2, $count);
         });
     }
 
@@ -215,18 +219,21 @@ public function testNotify() {
             /** @var \Amp\Postgres\Listener $listener */
             $listener = yield $this->connection->listen($channel);
 
-            yield $this->connection->notify($channel, '0');
-            yield $this->connection->notify($channel, '1');
+            asyncCall(function () use ($channel) {
+                yield $this->connection->notify($channel, '0');
+                yield $this->connection->notify($channel, '1');
+            });
 
             $count = 0;
-            Loop::defer(function () use (&$count, $listener) {
+            Loop::delay(100, function () use ($listener) {
                 $listener->unlisten();
-                $this->assertSame(2, $count);
             });
 
             while (yield $listener->advance()) {
                 $this->assertSame($listener->getCurrent()->payload, (string) $count++);
             }
+
+            $this->assertSame(2, $count);
         });
     }