Enabled unbuffered queries, but remove backpressure

src/PqConnection.php

@@ -27,9 +27,7 @@ final class PqConnection extends Connection {
         }
 
         $connection->nonblocking = true;
-
+        $connection->unbuffered = true;
-        // Disabling unbuffered results for now as there appears to be a bug when unbuffered results contain arrays.
-        // $connection->unbuffered = true;
 
         $deferred = new Deferred;
 

src/PqHandle.php

@@ -262,7 +262,7 @@ final class PqHandle implements Handle {
         }
 
         if (!$result) {
-            return null; // Connection closing, end result set.
+            throw new ConnectionException("Connection closed");
         }
 
         switch ($result->status) {
@@ -298,7 +298,7 @@ final class PqHandle implements Handle {
      * @throws \Amp\Postgres\FailureException
      */
     public function statementExecute(string $name, array $params): Promise {
-        \assert(isset($this->statements[$name]), "Named statement not found when deallocating");
+        \assert(isset($this->statements[$name]), "Named statement not found when executing");
 
         $statement = $this->statements[$name]->statement;
 

src/PqUnbufferedResultSet.php

@@ -2,7 +2,6 @@
 
 namespace Amp\Postgres;
 
-use Amp\Coroutine;
 use Amp\Producer;
 use Amp\Promise;
 use pq;
@@ -35,9 +34,8 @@ final class PqUnbufferedResultSet implements ResultSet, Operation {
             try {
                 do {
                     $result->autoConvert = pq\Result::CONV_SCALAR | pq\Result::CONV_ARRAY;
-                    $next = $fetch(); // Request next result before current is consumed.
                     yield $emit($result);
-                    $result = yield $next;
+                    $result = yield $fetch();
                 } while ($result instanceof pq\Result);
             } finally {
                 $queue->unreference();
@@ -45,12 +43,6 @@ final class PqUnbufferedResultSet implements ResultSet, Operation {
         });
     }
 
-    public function __destruct() {
-        if (!$this->queue->isReferenced()) { // Producer above did not complete, so consume remaining results.
-            Promise\rethrow(new Coroutine($this->dispose()));
-        }
-    }
-
     /**
      * {@inheritdoc}
      */
@@ -84,14 +76,6 @@ final class PqUnbufferedResultSet implements ResultSet, Operation {
         }
     }
 
-    private function dispose(): \Generator {
-        try {
-            while (yield $this->producer->advance()); // Discard unused result rows.
-        } catch (\Throwable $exception) {
-            // Ignore failure while discarding results.
-        }
-    }
-
     /**
      * @return int Number of fields (columns) in each result set.
      */

test/AbstractLinkTest.php

@@ -330,6 +330,31 @@ abstract class AbstractLinkTest extends TestCase {
         });
     }
 
+    public function testPrepareThenExecuteWithUnconsumedTupleResult() {
+        Loop::run(function () {
+            /** @var \Amp\Postgres\Statement $statement */
+            $statement = yield $this->connection->prepare("SELECT * FROM test");
+
+            /** @var \Amp\Postgres\ResultSet $result */
+            $result = yield $statement->execute();
+
+            $this->assertInstanceOf(ResultSet::class, $result);
+
+            /** @var \Amp\Postgres\ResultSet $result */
+            $result = yield $statement->execute();
+
+            $this->assertInstanceOf(ResultSet::class, $result);
+
+            $data = $this->getData();
+
+            for ($i = 0; yield $result->advance(ResultSet::FETCH_OBJECT); ++$i) {
+                $row = $result->getCurrent();
+                $this->assertSame($data[$i][0], $row->domain);
+                $this->assertSame($data[$i][1], $row->tld);
+            }
+        });
+    }
+
     public function testExecute() {
         Loop::run(function () {
             $data = $this->getData()[0];