Update for Stream to Pipeline changes

amphp · Aug 28, 2020 · dee4167 · dee4167
1 parent c092fdf
commit dee4167
Show file tree

Hide file tree

Showing 9 changed files with 18 additions and 94 deletions.
diff --git a/src/ConnectionListener.php b/src/ConnectionListener.php
@@ -2,12 +2,12 @@
 
 namespace Amp\Postgres;
 
+use Amp\Pipeline;
 use Amp\Promise;
-use Amp\Stream;
 
 final class ConnectionListener implements Listener
 {
-    /** @var Stream */
+    /** @var Pipeline */
     private $stream;
 
     /** @var string */
@@ -17,13 +17,13 @@ final class ConnectionListener implements Listener
     private $unlisten;
 
     /**
-     * @param Stream $stream  Stream emitting notificatons on the channel.
+     * @param Pipeline $pipeline Pipeline emitting notificatons on the channel.
      * @param string $channel Channel name.
      * @param callable(string $channel):  $unlisten Function invoked to unlisten from the channel.
      */
-    public function __construct(Stream $stream, string $channel, callable $unlisten)
+    public function __construct(Pipeline $pipeline, string $channel, callable $unlisten)
     {
-        $this->stream = $stream;
+        $this->stream = $pipeline;
         $this->channel = $channel;
         $this->unlisten = $unlisten;
     }
@@ -52,22 +52,6 @@ public function dispose()
         $this->unlisten();
     }
 
-    /**
-     * @inheritDoc
-     */
-    public function onDisposal(callable $onDisposal): void
-    {
-        $this->stream->onDisposal($onDisposal);
-    }
-
-    /**
-     * @inheritDoc
-     */
-    public function onCompletion(callable $onCompletion): void
-    {
-        $this->stream->onCompletion($onCompletion);
-    }
-
     /**
      * @return string Channel name.
      */

diff --git a/src/Listener.php b/src/Listener.php
@@ -2,10 +2,10 @@
 
 namespace Amp\Postgres;
 
+use Amp\Pipeline;
 use Amp\Promise;
-use Amp\Stream;
 
-interface Listener extends Stream
+interface Listener extends Pipeline
 {
     /**
      * @return Promise<Notification|null>

diff --git a/src/PgSqlHandle.php b/src/PgSqlHandle.php
@@ -4,13 +4,13 @@
 
 use Amp\Deferred;
 use Amp\Loop;
+use Amp\PipelineSource;
 use Amp\Promise;
 use Amp\Sql\Common\CommandResult;
 use Amp\Sql\ConnectionException;
 use Amp\Sql\FailureException;
 use Amp\Sql\QueryError;
 use Amp\Sql\Result;
-use Amp\StreamSource;
 use Amp\Struct;
 use Amp\Success;
 use function Amp\call;
@@ -44,7 +44,7 @@ final class PgSqlHandle implements Handle
     /** @var string */
     private $await;
 
-    /** @var StreamSource[] */
+    /** @var PipelineSource[] */
     private $listeners = [];
 
     /** @var object[] Anonymous class using Struct trait. */
@@ -463,7 +463,7 @@ public function listen(string $channel): Promise
                 throw new QueryError(\sprintf("Already listening on channel '%s'", $channel));
             }
 
-            $this->listeners[$channel] = $source = new StreamSource;
+            $this->listeners[$channel] = $source = new PipelineSource;
 
             try {
                 yield $this->query(\sprintf("LISTEN %s", $this->quoteName($channel)));
@@ -473,7 +473,7 @@ public function listen(string $channel): Promise
             }
 
             Loop::enable($this->poll);
-            return new ConnectionListener($source->stream(), $channel, \Closure::fromCallable([$this, 'unlisten']));
+            return new ConnectionListener($source->pipe(), $channel, \Closure::fromCallable([$this, 'unlisten']));
         });
     }
 

diff --git a/src/PgSqlResultSet.php b/src/PgSqlResultSet.php
@@ -55,7 +55,6 @@ public function __construct($handle, Promise $nextResult)
                 \pg_free_result($handle);
             }
         });
-        $this->generator->getReturn(); // Force generator to start execution.
     }
 
     /**
@@ -74,22 +73,6 @@ public function dispose(): void
         $this->generator->dispose();
     }
 
-    /**
-     * @inheritDoc
-     */
-    public function onDisposal(callable $onDisposal): void
-    {
-        $this->generator->onDisposal($onDisposal);
-    }
-
-    /**
-     * @inheritDoc
-     */
-    public function onCompletion(callable $onCompletion): void
-    {
-        $this->generator->onCompletion($onCompletion);
-    }
-
     /**
      * @inheritDoc
      */

diff --git a/src/PooledListener.php b/src/PooledListener.php
@@ -40,16 +40,6 @@ public function dispose(): void
         $this->listener->dispose();
     }
 
-    public function onDisposal(callable $onDisposal): void
-    {
-        $this->listener->onDisposal($onDisposal);
-    }
-
-    public function onCompletion(callable $onCompletion): void
-    {
-        $this->listener->onCompletion($onCompletion);
-    }
-
     public function getChannel(): string
     {
         return $this->listener->getChannel();

diff --git a/src/PqBufferedResultSet.php b/src/PqBufferedResultSet.php
@@ -53,22 +53,6 @@ public function dispose(): void
         $this->generator->dispose();
     }
 
-    /**
-     * @inheritDoc
-     */
-    public function onDisposal(callable $onDisposal): void
-    {
-        $this->generator->onDisposal($onDisposal);
-    }
-
-    /**
-     * @inheritDoc
-     */
-    public function onCompletion(callable $onCompletion): void
-    {
-        $this->generator->onCompletion($onCompletion);
-    }
-
     /**
      * @inheritDoc
      */

diff --git a/src/PqHandle.php b/src/PqHandle.php
@@ -5,13 +5,13 @@
 use Amp\Coroutine;
 use Amp\Deferred;
 use Amp\Loop;
+use Amp\PipelineSource;
 use Amp\Promise;
 use Amp\Sql\Common\CommandResult;
 use Amp\Sql\ConnectionException;
 use Amp\Sql\FailureException;
 use Amp\Sql\QueryError;
 use Amp\Sql\Result;
-use Amp\StreamSource;
 use Amp\Struct;
 use Amp\Success;
 use pq;
@@ -34,7 +34,7 @@ final class PqHandle implements Handle
     /** @var string */
     private $await;
 
-    /** @var StreamSource[] */
+    /** @var PipelineSource[] */
     private $listeners;
 
     /** @var object[] Anonymous class using Struct trait. */
@@ -469,7 +469,7 @@ public function listen(string $channel): Promise
                 throw new QueryError(\sprintf("Already listening on channel '%s'", $channel));
             }
 
-            $this->listeners[$channel] = $source = new StreamSource;
+            $this->listeners[$channel] = $source = new PipelineSource;
 
             try {
                 yield from $this->send(
@@ -490,7 +490,7 @@ static function (string $channel, string $message, int $pid) use ($source) {
             }
 
             Loop::enable($this->poll);
-            return new ConnectionListener($source->stream(), $channel, \Closure::fromCallable([$this, 'unlisten']));
+            return new ConnectionListener($source->pipe(), $channel, \Closure::fromCallable([$this, 'unlisten']));
         });
     }
 

diff --git a/src/PqUnbufferedResultSet.php b/src/PqUnbufferedResultSet.php
@@ -39,7 +39,6 @@ public function __construct(callable $fetch, pq\Result $result, Promise $nextRes
                 }
             }
         });
-        $this->generator->getReturn(); // Force generator to start execution.
     }
 
     public function getNextResult(): Promise
@@ -63,22 +62,6 @@ public function dispose(): void
         $this->generator->dispose();
     }
 
-    /**
-     * @inheritDoc
-     */
-    public function onDisposal(callable $onDisposal): void
-    {
-        $this->generator->onDisposal($onDisposal);
-    }
-
-    /**
-     * @inheritDoc
-     */
-    public function onCompletion(callable $onCompletion): void
-    {
-        $this->generator->onCompletion($onCompletion);
-    }
-
     /**
      * @inheritDoc
      */

diff --git a/test/AbstractLinkTest.php b/test/AbstractLinkTest.php
@@ -6,6 +6,7 @@
 use Amp\Delayed;
 use Amp\Loop;
 use Amp\PHPUnit\AsyncTestCase;
+use Amp\Pipeline;
 use Amp\Postgres\Link;
 use Amp\Postgres\Listener;
 use Amp\Postgres\QueryExecutionError;
@@ -16,7 +17,6 @@
 use Amp\Sql\Statement;
 use Amp\Sql\Transaction as SqlTransaction;
 use Amp\Sql\TransactionError;
-use Amp\Stream;
 
 abstract class AbstractLinkTest extends AsyncTestCase
 {
@@ -433,8 +433,8 @@ public function testPrepareSimilarQueryReturnsDifferentStatements(): \Generator
 
         $results = [];
 
-        $results[] = yield Stream\toArray(yield $statement1->execute([$data[0]]));
-        $results[] = yield Stream\toArray(yield $statement2->execute(['domain' => $data[0]]));
+        $results[] = yield Pipeline\toArray(yield $statement1->execute([$data[0]]));
+        $results[] = yield Pipeline\toArray(yield $statement2->execute(['domain' => $data[0]]));
 
         foreach ($results as $result) {
             /** @var Result $result */