Something went wrong. Try again.
Build Reactive Signals for Bluesky's AT Protocol Firehose in Laravel
Something went wrong. Try again.
4.1 kB · 176 lines
PHP
at dev
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177<?php
namespace SocialDept\AtpSignals\Support;
use Ratchet\Client\Connector;use Ratchet\Client\WebSocket;use Ratchet\RFC6455\Messaging\MessageInterface;use React\EventLoop\Loop;use React\EventLoop\LoopInterface;use React\Promise\PromiseInterface;
/** * WebSocket connection wrapper for AT Protocol Jetstream. */class WebSocketConnection{ protected ?WebSocket $connection = null; protected LoopInterface $loop; protected bool $connected = false; protected ?\Closure $onMessage = null; protected ?\Closure $onClose = null; protected ?\Closure $onError = null;
public function __construct(?LoopInterface $loop = null) { $this->loop = $loop ?? Loop::get(); }
/** * Connect to a WebSocket endpoint. */ public function connect(string $url): PromiseInterface { $connector = new Connector($this->loop);
return $connector($url)->then( function (WebSocket $conn) { $this->connection = $conn; $this->connected = true;
// Register event handlers $conn->on('message', function (MessageInterface $msg) { if ($this->onMessage) { ($this->onMessage)($msg->getPayload()); } });
$conn->on('close', function ($code = null, $reason = null) { $this->connected = false; if ($this->onClose) { ($this->onClose)($code, $reason); } });
$conn->on('error', function (\Exception $e) { if ($this->onError) { ($this->onError)($e); } });
return $conn; }, function (\Exception $e) { if ($this->onError) { ($this->onError)($e); }
throw $e; } ); }
/** * Send a message through the WebSocket connection. */ public function send(string $message): bool { if (! $this->connected || ! $this->connection) { return false; }
try { $this->connection->send($message);
return true; } catch (\Exception $e) { if ($this->onError) { ($this->onError)($e); }
return false; } }
/** * Close the WebSocket connection. */ public function close(int $code = 1000, string $reason = ''): void { if ($this->connection && $this->connected) { $this->connection->close($code, $reason); $this->connected = false; } }
/** * Check if the connection is currently active. */ public function isConnected(): bool { return $this->connected; }
/** * Set the message handler callback. */ public function onMessage(callable $callback): self { $this->onMessage = $callback(...);
return $this; }
/** * Set the close handler callback. */ public function onClose(callable $callback): self { $this->onClose = $callback(...);
return $this; }
/** * Set the error handler callback. */ public function onError(callable $callback): self { $this->onError = $callback(...);
return $this; }
/** * Get the event loop instance. */ public function getLoop(): LoopInterface { return $this->loop; }
/** * Start the event loop (blocking). */ public function run(): void { $this->loop->run(); }
/** * Stop the event loop. */ public function stop(): void { $this->close(); $this->loop->stop(); }
/** * Get the underlying WebSocket connection. */ public function getConnection(): ?WebSocket { return $this->connection; }}