Update for Stream onDisposal() and onCompletion()

src/CommandResult.php

@@ -2,8 +2,7 @@
 
 namespace Amp\Sql\Common;
 
-use Amp\DisposedException;
-use Amp\Failure;
+use Amp\Loop;
 use Amp\Promise;
 use Amp\Sql\Result;
 use Amp\Success;
@@ -33,7 +32,23 @@ public function continue(): Promise
 
     public function dispose(): void
     {
-        $this->promise = new Failure(new DisposedException);
+        // No-op
+    }
+
+    public function onDisposal(callable $onDisposal): void
+    {
+        // No-op, result is complete on creation
+    }
+
+    public function onCompletion(callable $onCompletion): void
+    {
+        try {
+            $onCompletion(null);
+        } catch (\Throwable $e) {
+            Loop::defer(static function () use ($e): void {
+                throw $e;
+            });
+        }
     }
 
     public function getNextResult(): Promise

src/PooledResult.php

@@ -72,6 +72,16 @@ public function dispose(): void
         }
     }
 
+    public function onDisposal(callable $onDisposal): void
+    {
+        $this->result->onDisposal($onDisposal);
+    }
+
+    public function onCompletion(callable $onCompletion): void
+    {
+        $this->result->onCompletion($onCompletion);
+    }
+
     public function getRowCount(): ?int
     {
         return $this->result->getRowCount();