reactphp-kernel/src/Support/ConnectionRegistry.php

<?php

declare(strict_types=1);

namespace Blax\ReactPhpKernel\Support;

/**
 * A generic, in-memory registry of active connections/peers keyed by string id.
 *
 * Shared plumbing for presence + broadcast in protocol servers — WS channel
 * members, WebRTC peers in a room, etc. It is transport-agnostic: values are
 * arbitrary objects (a socket connection, a peer wrapper, ...), so the same
 * registry serves every protocol on the kernel.
 */
final class ConnectionRegistry implements \Countable
{
    /** @var array<string,object> */
    private array $items = [];

    /** @var array<callable> */
    private array $onAdd = [];

    /** @var array<callable> */
    private array $onRemove = [];

    /** Register (or replace) a connection under $id. Fires onAdd listeners. */
    public function add(string $id, object $connection): void
    {
        $this->items[$id] = $connection;

        foreach ($this->onAdd as $listener) {
            $listener($id, $connection);
        }
    }

    /** Remove a connection by id (no-op if absent). Fires onRemove listeners. */
    public function remove(string $id): void
    {
        if (! isset($this->items[$id])) {
            return;
        }

        $connection = $this->items[$id];
        unset($this->items[$id]);

        foreach ($this->onRemove as $listener) {
            $listener($id, $connection);
        }
    }

    public function has(string $id): bool
    {
        return isset($this->items[$id]);
    }

    public function get(string $id): ?object
    {
        return $this->items[$id] ?? null;
    }

    /** @return array<string,object> */
    public function all(): array
    {
        return $this->items;
    }

    /** @return array<int,string> */
    public function ids(): array
    {
        return array_keys($this->items);
    }

    public function count(): int
    {
        return count($this->items);
    }

    /**
     * Apply a callback to every connection — the broadcast primitive.
     * The callback receives (object $connection, string $id).
     */
    public function each(callable $callback): void
    {
        foreach ($this->items as $id => $connection) {
            $callback($connection, $id);
        }
    }

    /** Listen for additions: function(string $id, object $connection): void. */
    public function onAdd(callable $listener): void
    {
        $this->onAdd[] = $listener;
    }

    /** Listen for removals: function(string $id, object $connection): void. */
    public function onRemove(callable $listener): void
    {
        $this->onRemove[] = $listener;
    }

    /** Remove every connection (firing onRemove for each). */
    public function clear(): void
    {
        foreach (array_keys($this->items) as $id) {
            $this->remove($id);
        }
    }
}