diff --git a/src/Future.php b/src/Future.php index 313b5980..4c6d8b2c 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 is always queued for execution on the event loop, even 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..8057f33f 100644 --- a/test/Future/FutureTest.php +++ b/test/Future/FutureTest.php @@ -275,6 +275,90 @@ 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 + + self::assertSame(1, $future->await()); + } + + public function testUnsubscribeAfterCompletionDoesNotCancelQueuedCallback(): void + { + $called = false; + $future = Future::complete(1); + $id = $future->subscribe(static function () use (&$called): void { + $called = true; + }); + + $future->unsubscribe($id); + self::assertFalse($called); + + delay(0); // tick event loop + + self::assertTrue($called); + } + + public function testThrowingSubscriptionCallbackEndsUpInEventLoop(): void + { + EventLoop::setErrorHandler(static function (\Throwable $exception) use (&$reason): void { + $reason = $exception; + }); + + Future::complete()->subscribe(static function (): void { + throw new \Exception('subscription callback failed'); + }); + + delay(0); // tick event loop + + self::assertInstanceOf(\Exception::class, $reason); + self::assertSame('subscription callback failed', $reason->getMessage()); + } + public function testMapWithCompleteFuture(): void { $future = Future::complete(1);