Consume further results after query error

Fixes #32.

Author: trowski

src/PgSqlHandle.php

@@ -266,13 +266,17 @@ private function createResult($result, string $sql)
                 foreach (self::DIAGNOSTIC_CODES as $fieldCode => $desciption) {
                     $diagnostics[$desciption] = \pg_result_error_field($result, $fieldCode);
                 }
-                throw new QueryExecutionError(\pg_result_error($result), $diagnostics, null, $sql);
+                $message = \pg_result_error($result);
+                while (\pg_connection_busy($this->handle) && \pg_get_result($this->handle));
+                throw new QueryExecutionError($message, $diagnostics, null, $sql);
 
             case \PGSQL_BAD_RESPONSE:
+                $this->close();
                 throw new FailureException(\pg_result_error($result));
 
             default:
                 // @codeCoverageIgnoreStart
+                $this->close();
                 throw new FailureException("Unknown result status");
                 // @codeCoverageIgnoreEnd
         }

src/PqHandle.php

@@ -233,12 +233,15 @@ private function send(?string $sql, callable $method, ...$args): \Generator
 
             case pq\Result::NONFATAL_ERROR:
             case pq\Result::FATAL_ERROR:
+                while ($this->handle->busy && $this->handle->getResult());
                 throw new QueryExecutionError($result->errorMessage, $result->diag, null, $sql ?? '');
 
             case pq\Result::BAD_RESPONSE:
+                $this->close();
                 throw new FailureException($result->errorMessage);
 
             default:
+                $this->close();
                 throw new FailureException("Unknown result status");
         }
     }
@@ -274,6 +277,7 @@ private function fetch(): \Generator
                 return $result;
 
             default:
+                $this->close();
                 throw new FailureException($result->errorMessage);
         }
     }

test/AbstractLinkTest.php

@@ -682,4 +682,18 @@ public function testListenOnSameChannel(): Promise
         $channel = "test";
         return Promise\all([$this->connection->listen($channel), $this->connection->listen($channel)]);
     }
+
+    public function testQueryAfterErroredQuery(): \Generator
+    {
+        try {
+            $result = yield $this->connection->query("INSERT INTO test (domain, tld) VALUES ('github', 'com')");
+        } catch (QueryExecutionError $exception) {
+            // Expected exception due to duplicate key.
+        }
+
+        /** @var CommandResult $result */
+        $result = yield $this->connection->query("INSERT INTO test (domain, tld) VALUES ('gitlab', 'com')");
+
+        $this->assertSame(1, $result->getAffectedRowCount());
+    }
 }