Restore backpressure to unbuffered results

Author: trowski

src/PqUnbufferedResultSet.php

@@ -5,6 +5,7 @@
 use Amp\Producer;
 use Amp\Promise;
 use pq;
+use function Amp\asyncCall;
 
 final class PqUnbufferedResultSet implements ResultSet
 {
@@ -17,6 +18,9 @@ final class PqUnbufferedResultSet implements ResultSet
     /** @var array|object Last row emitted. */
     private $currentRow;
 
+    /** @var bool */
+    private $destroyed = false;
+
     /**
      * @param callable():  $fetch Function to fetch next result row.
      * @param \pq\Result $result PostgreSQL result object.
@@ -25,19 +29,36 @@ public function __construct(callable $fetch, pq\Result $result, callable $releas
     {
         $this->numCols = $result->numCols;
 
-        $this->producer = new Producer(static function (callable $emit) use ($release, $result, $fetch) {
+        $this->producer = new Producer(static function (callable $emit) use (&$destroyed, $release, $result, $fetch) {
             try {
                 do {
                     $result->autoConvert = pq\Result::CONV_SCALAR | pq\Result::CONV_ARRAY;
-                    $emit($result);
-                    $result = yield $fetch();
+                    $next = $fetch();
+                    yield $emit($result);
+                    $result = yield $next;
                 } while ($result instanceof pq\Result);
             } finally {
+                $destroyed = true;
                 $release();
             }
         });
     }
 
+    public function __destruct()
+    {
+        if ($this->destroyed) {
+            return;
+        }
+
+        asyncCall(function () {
+            try {
+                while (yield $this->producer->advance());
+            } catch (\Throwable $exception) {
+                // Ignore iterator failure when destroying.
+            }
+        });
+    }
+
     /**
      * {@inheritdoc}
      */
@@ -58,7 +79,7 @@ public function getCurrent(): array
         }
 
         $result = $this->producer->getCurrent();
-        \assert($result instanceof \pq\Result);
+        \assert($result instanceof pq\Result);
 
         return $this->currentRow = $result->fetchRow(pq\Result::FETCH_ASSOC);
     }