diff --git a/src/Future.php b/src/Future.php index 313b5980..ecbe65a1 100644 --- a/src/Future.php +++ b/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. diff --git a/test/Future/FutureTest.php b/test/Future/FutureTest.php index 613a4f43..254f281b 100644 --- a/test/Future/FutureTest.php +++ b/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);