Building JavaScript-Like `async` / `await` in PHP with Swoole and OpenSwoole

use Small\SwoolePatterns\Async\Async;
use function Small\SwoolePatterns\Async\async;
use function Small\SwoolePatterns\Async\await;

$result = Async::main(function (): array {
    $user = async(fn() => loadUser(42));
    $orders = async(fn() => loadOrders(42));

    return [
        'user' => await($user),
        'orders' => await($orders),
    ];
});

Application
    │
    ▼
Async::main()
    │
    ▼
Runtime::run()
    │
    ├───────────────┐
    ▼               ▼
 async()           async()
    │               │
    ▼               ▼
Future<T>       Future<T>
    │               │
    ▼               ▼
Coroutine       Coroutine
    │               │
    └───────┬───────┘
            ▼
          await()
            │
            ▼
       Channel::pop()

namespace Small\SwoolePatterns\Async;

/**
 * @template-covariant T
 */
interface AwaitableInterface
{
    /**
     * @return T
     */
    public function await(?float $timeout = null): mixed;
}

HTTP responses
database queries
timers
signals
pool acquisitions
message queue operations

enum FutureState
{
    case pending;
    case fulfilled;
    case rejected;
}

private FutureState $state = FutureState::pending;
private mixed $value = null;
private ?Throwable $exception = null;
private Channel $signal;

$future = async(fn() => expensiveOperation());

async(fn() => consume(await($future)));
async(fn() => logResult(await($future)));
async(fn() => cache(await($future)));

private function settle(
    FutureState $state,
    mixed $value,
    ?Throwable $exception,
): void {
    if ($this->state !== FutureState::pending) {
        return;
    }

    $this->state = $state;
    $this->value = $value;
    $this->exception = $exception;

    $this->signal->close();
}

Future pending
      │
      ├── coroutine A waiting
      ├── coroutine B waiting
      └── coroutine C waiting
             │
             ▼
        Future settles
             │
             ▼
        Channel closes
             │
      ┌──────┼──────┐
      ▼      ▼      ▼
      A      B      C

public function await(?float $timeout = null): mixed
{
    if ($timeout !== null && $timeout < 0) {
        throw new InvalidArgumentException(
            'Future timeout must be null or greater than or equal to 0.'
        );
    }

    if ($this->isPending()) {
        if (!Runtime::isInCoroutine()) {
            throw new AsyncException(
                'A pending Future can only be awaited inside Async::main() or Runtime::run().'
            );
        }

        $this->signal->pop($timeout ?? -1);

        if ($this->isPending()) {
            throw new TimeoutException(
                'Future did not complete before the timeout.'
            );
        }
    }

    if ($this->state === FutureState::rejected) {
        throw $this->exception;
    }

    return $this->value;
}

$value = await($future);

check state
return value

public static function start(
    Closure $callback,
    mixed ...$arguments,
): Future {
    $future = new Future(
        function (Closure $resolve, Closure $reject)
            use ($callback, $arguments): void {

            Runtime::create(
                function () use (
                    $callback,
                    $arguments,
                    $resolve,
                    $reject
                ): void {
                    try {
                        $result = $callback(...$arguments);
                        $resolve($result);
                    } catch (Throwable $exception) {
                        $reject($exception);
                    }
                }
            );
        }
    );

    return $future;
}

$future = async(fn() => queryDatabase());

$a = async(fn() => operationA());
$b = async(fn() => operationB());

$resultA = await($a);
$resultB = await($b);

Async::main(function () {
    // async code
});

public static function main(Closure $callback): mixed
{
    return Runtime::run(
        static function () use ($callback): mixed {
            $result = $callback();

            return $result instanceof AwaitableInterface
                ? $result->await()
                : $result;
        }
    );
}

$result = $callback(...$arguments);

if ($result instanceof AwaitableInterface) {
    $result = $result->await();
}

$resolve($result);

$future = async(function () {
    throw new RuntimeException('Database unavailable');
});

try {
    $result = await($future);
} catch (RuntimeException $exception) {
    echo $exception->getMessage();
}

$result = await(
    Async::all([
        'user' => async(fn() => loadUser()),
        'orders' => async(fn() => loadOrders()),
        'stock' => async(fn() => loadStock()),
    ])
);

[
    'user' => $user,
    'orders' => $orders,
    'stock' => $stock,
]

try {
    $value = await($future, timeout: 1.5);
} catch (TimeoutException) {
    // operation did not finish in 1.5 seconds
}

try {
    await($future, timeout: 0.1);
} catch (TimeoutException) {
    // continue doing something else
}

$result = await($future);

$future
    ->then(...)
    ->then(...)
    ->catch(...);

try {
    $user = await($userFuture);
    $orders = await($orderFuture);

    return buildResponse($user, $orders);
} catch (Throwable $exception) {
    // ...
}

$result = await(
    async(fn() => operation())
);

1 Future
1 coroutine
1 Channel
1 stored value
1 optional exception

Swoole 6.2.x
OpenSwoole 26.2.x

167 tests
441 assertions

Classes: 100%
Methods: 100%
Lines:   100%

$run ??= method_exists(Swoole\Coroutine::class, 'run') ? [Swoole\Coroutine::class, 'run'] : 'Swoole\\Coroutine\\run';

$result = Async::main(function () {

    $product = async(fn() => loadProduct());
    $stock = async(fn() => loadStock());
    $prices = async(fn() => loadPrices());

    return [
        'product' => await($product),
        'stock' => await($stock),
        'prices' => await($prices),
    ];
});

const productPromise = loadProduct();
const stockPromise = loadStock();
const pricesPromise = loadPrices();

const result = {
    product: await productPromise,
    stock: await stockPromise,
    prices: await pricesPromise
};

start operation
receive eventual result
continue doing work
await when needed
propagate exception normally