Add Future completion subscriptions

src/Future.php

@@ -137,6 +137,31 @@ public function ignore(): self
         return $this;
     }
 
+    /**
+     * Subscribes a callback to be invoked when this Future completes or errors.
+     *
+     * The callback might be invoked immediately if this Future has already completed. The callback must not suspend.
+     * Any unhandled exceptions will be thrown into the event loop.
+     *
+     * @param \Closure(?\Throwable, mixed): void $callback Callback to be invoked on error or successful completion.
+     *
+     * @return string Identifier that can be used to cancel the subscription.
+     */
+    public function subscribe(\Closure $callback): string
+    {
+        return $this->state->subscribe($callback);
+    }
+
+    /**
+     * Unsubscribes a previously registered callback.
+     *
+     * The callback might still be invoked if the Future has already completed.
+     */
+    public function unsubscribe(string $id): void
+    {
+        $this->state->unsubscribe($id);
+    }
+
     /**
      * Attaches a callback that is invoked if this future completes. The returned future is completed with the return
      * value of the callback, or errors with an exception thrown from the callback.

test/Future/FutureTest.php

@@ -275,6 +275,56 @@ public function testIgnoreUnhandledErrorFromFutureError(): void
         delay(0); // tick event loop
     }
 
+    public function testSubscribeWithCompletedFuture(): void
+    {
+        Future::complete(1)->subscribe(function (?\Throwable $error, mixed $value): void {
+            self::assertNull($error);
+            self::assertSame(1, $value);
+        });
+
+        delay(0); // tick event loop
+    }
+
+    public function testSubscribeWithErroredFuture(): void
+    {
+        $exception = new \Exception();
+
+        Future::error($exception)->subscribe(function (?\Throwable $error, mixed $value) use ($exception): void {
+            self::assertSame($exception, $error);
+            self::assertNull($value);
+        });
+
+        delay(0); // tick event loop
+    }
+
+    public function testSubscribeWithPendingFuture(): void
+    {
+        $deferred = new DeferredFuture;
+        $future = $deferred->getFuture();
+        $future->subscribe(function (?\Throwable $error, mixed $value): void {
+            self::assertNull($error);
+            self::assertSame(1, $value);
+        });
+
+        $deferred->complete(1);
+
+        delay(0); // tick event loop
+    }
+
+    public function testUnsubscribe(): void
+    {
+        $deferred = new DeferredFuture;
+        $future = $deferred->getFuture();
+        $id = $future->subscribe(function (): void {
+            self::fail('Callback has been called');
+        });
+
+        $future->unsubscribe($id);
+        $deferred->complete(1);
+
+        delay(0); // tick event loop
+    }
+
     public function testMapWithCompleteFuture(): void
     {
         $future = Future::complete(1);