Sincronización de Datos con WebSockets y Reactivos Observables en PHP

En aplicaciones modernas, la actualización de datos en tiempo real es crucial. Tradicionalmente, esto se ha abordado con polling o long-polling, que pueden ser ineficientes. Una alternativa superior es usar WebSockets para una comunicación bidireccional persistente y observables reactivos para manejar los flujos de datos asíncronos. Este post explora cómo combinar estas tecnologías en PHP para una sincronización de datos eficiente.

Usaremos Ratchet, una biblioteca de WebSockets para PHP, y un observable reactivo simple implementado con generadores para demostrar el concepto. La idea es que los cambios en una fuente de datos (por ejemplo, una base de datos) se detecten y se transmitan a través de un observable, que luego notifica a todos los clientes WebSocket conectados.

<?php

use Ratchet\MessageComponentInterface;
use Ratchet\ConnectionInterface;
use React\EventLoop\Factory;
use Ratchet\Http\HttpServer;
use Ratchet\WebSocket\WsServer;
use Ratchet\Server\IoServer;

// Simple Observable
class Observable {
    private $generator;

    public function __construct(Generator $generator) {
        $this->generator = $generator;
    }

    public function subscribe(callable $next) {
        foreach ($this->generator as $value) {
            $next($value);
        }
    }
}

// WebSocket Server
class DataSyncServer implements MessageComponentInterface {
    protected $clients;
    protected $dataObservable;

    public function __construct(Observable $dataObservable) {
        $this->clients = new \SplObjectStorage;
        $this->dataObservable = $dataObservable;

        $this->dataObservable->subscribe(function ($data) {
            foreach ($this->clients as $client) {
                $client->send(json_encode($data));
            }
        });
    }

    public function onOpen(ConnectionInterface $conn) {
        $this->clients->attach($conn);
        echo "New connection! ({$conn->resourceId})\n";
    }

    public function onMessage(ConnectionInterface $from, $msg) {
        // No se usa en este ejemplo
    }

    public function onClose(ConnectionInterface $conn) {
        $this->clients->detach($conn);
        echo "Connection {$conn->resourceId} has disconnected\n";
    }

    public function onError(ConnectionInterface $conn, \Exception $e) {
        echo "An error has occurred: {$e->getMessage()}\n";
        $conn->close();
    }
}

// Fuente de datos simulada (reemplazar con acceso a base de datos real)
function mockDataSource(): Generator {
    $i = 0;
    while (true) {
        yield ['value' => rand(1, 100), 'timestamp' => time()];
        sleep(1); // Simula un cambio en la fuente de datos cada segundo
        $i++;
        if($i > 10) {
            break;
        }
    }
}


// Configuración del servidor
$loop   = Factory::create();
$dataObservable = new Observable(mockDataSource()); // Instancia del Observable
$server = IoServer::factory(
    new HttpServer(
        new WsServer(
            new DataSyncServer($dataObservable) // Pasa el Observable al servidor
        )
    ),
    8080,
    '0.0.0.0',
    $loop
);

$loop->run();

Este código establece un servidor WebSocket que transmite datos generados por un "observable". El observable, en este caso, es una simulación, pero en un escenario real, podría estar conectado a una base de datos u otra fuente de datos. Cada vez que la fuente de datos cambia (simulado por sleep(1)), el observable notifica al servidor, que a su vez envía los datos actualizados a todos los clientes WebSocket conectados.

Para probarlo, necesitarás un cliente WebSocket. Puedes usar un cliente WebSocket en JavaScript en una página HTML. Recuerda que este es un ejemplo simplificado. En una aplicación real, necesitarás manejar la autenticación, la autorización y la persistencia de datos de forma segura.