tipo: modificación del cpanel y gitignore de la carpeta vendor

This commit is contained in:
2026-07-07 11:35:09 -06:00
parent a0c91dc567
commit 676ef0d506
8510 changed files with 1132803 additions and 4 deletions
+133
View File
@@ -0,0 +1,133 @@
<?php
namespace Laravel\Reverb;
class Application
{
/**
* Create a new application instance.
*/
public function __construct(
protected string $id,
protected string $key,
protected string $secret,
protected int $pingInterval,
protected int $activityTimeout,
protected array $allowedOrigins,
protected int $maxMessageSize,
protected ?int $maxConnections = null,
protected string $acceptClientEventsFrom = 'members',
protected array $options = [],
) {
//
}
/**
* Get the application ID.
*/
public function id(): string
{
return $this->id;
}
/**
* Get the application key.
*/
public function key(): string
{
return $this->key;
}
/**
* Get the application secret.
*/
public function secret(): string
{
return $this->secret;
}
/**
* Get the allowed origins.
*
* @return array<int, string>
*/
public function allowedOrigins(): array
{
return $this->allowedOrigins;
}
/**
* Get the client ping interval in seconds.
*/
public function pingInterval(): int
{
return $this->pingInterval;
}
/**
* Get the activity timeout in seconds.
*/
public function activityTimeout(): int
{
return $this->activityTimeout;
}
/**
* Get the maximum connections allowed for the application.
*/
public function maxConnections(): ?int
{
return $this->maxConnections;
}
/**
* Determine if the application has a maximum connection limit.
*/
public function hasMaxConnectionLimit(): bool
{
return $this->maxConnections !== null;
}
/**
* Get the maximum message size allowed from the client.
*/
public function maxMessageSize(): int
{
return $this->maxMessageSize;
}
/**
* Get who client events are accepted from for the application - either "all", "members", or "none".
*/
public function acceptClientEventsFrom(): string
{
return $this->acceptClientEventsFrom;
}
/**
* Get the application options.
*/
public function options(): ?array
{
return $this->options;
}
/**
* Convert the application to an array.
*
* @return array<string, mixed>
*/
public function toArray(): array
{
return [
'app_id' => $this->id,
'key' => $this->key,
'secret' => $this->secret,
'ping_interval' => $this->pingInterval,
'activity_timeout' => $this->activityTimeout,
'allowed_origins' => $this->allowedOrigins,
'max_message_size' => $this->maxMessageSize,
'options' => $this->options,
];
}
}
+26
View File
@@ -0,0 +1,26 @@
<?php
namespace Laravel\Reverb;
use Illuminate\Support\Manager;
class ApplicationManager extends Manager
{
/**
* Create an instance of the configuration driver.
*/
public function createConfigDriver(): ConfigApplicationProvider
{
return new ConfigApplicationProvider(
collect($this->config->get('reverb.apps.apps', []))
);
}
/**
* Get the default driver name.
*/
public function getDefaultDriver(): string
{
return $this->config->get('reverb.apps.provider', 'config');
}
}
@@ -0,0 +1,33 @@
<?php
namespace Laravel\Reverb;
use Illuminate\Contracts\Support\DeferrableProvider;
use Illuminate\Support\ServiceProvider;
use Laravel\Reverb\Contracts\ApplicationProvider;
class ApplicationManagerServiceProvider extends ServiceProvider implements DeferrableProvider
{
/**
* Register any application services.
*/
public function register(): void
{
$this->app->singleton(ApplicationManager::class);
$this->app->bind(
ApplicationProvider::class,
fn ($app) => $app->make(ApplicationManager::class)->driver()
);
}
/**
* Get the services provided by the provider.
*
* @return array<int, class-string>
*/
public function provides(): array
{
return [ApplicationManager::class, ApplicationProvider::class];
}
}
+67
View File
@@ -0,0 +1,67 @@
<?php
namespace Laravel\Reverb;
class Certificate
{
/**
* Determine if the certificate exists.
*/
public static function exists(string $url): bool
{
return static::resolve($url) !== null;
}
/**
* Resolve the certificate and key for the given URL.
*
* @return array<int, string>|null
*/
public static function resolve(string $url): ?array
{
$host = parse_url($url, PHP_URL_HOST) ?: $url;
$certificate = $host.'.crt';
$key = $host.'.key';
foreach (static::paths() as $path) {
if (file_exists($path.$certificate) && file_exists($path.$key)) {
return [$path.$certificate, $path.$key];
}
}
return null;
}
/**
* Get the certificate paths.
*
* @return array<int, string>
*/
public static function paths(): array
{
return [
static::herdPath(),
static::valetPath(),
];
}
/**
* Get the Herd certificate path.
*/
public static function herdPath(): string
{
if (PHP_OS_FAMILY === 'Windows') {
return implode(DIRECTORY_SEPARATOR, [getenv('USERPROFILE') ?: $_SERVER['HOME'] ?? '', '.config', 'herd', 'config', 'valet', 'Certificates', '']);
}
return implode(DIRECTORY_SEPARATOR, [$_SERVER['HOME'] ?? '', 'Library', 'Application Support', 'Herd', 'config', 'valet', 'Certificates', '']);
}
/**
* Get the Valet certificate path.
*/
public static function valetPath(): string
{
return implode(DIRECTORY_SEPARATOR, [$_SERVER['HOME'] ?? '', '.config', 'valet', 'Certificates', '']);
}
}
@@ -0,0 +1,14 @@
<?php
namespace Laravel\Reverb\Concerns;
trait GeneratesIdentifiers
{
/**
* Generate a Pusher-compatible socket ID.
*/
protected function generateId(): string
{
return sprintf('%d.%d', random_int(1, 1_000_000_000), random_int(1, 1_000_000_000));
}
}
@@ -0,0 +1,18 @@
<?php
namespace Laravel\Reverb\Concerns;
use Laravel\Reverb\Application;
trait InteractsWithApplications
{
/**
* Set the application the channel manager should be scoped to.
*/
public function for(Application $application): self
{
$this->application = $application;
return $this;
}
}
@@ -0,0 +1,38 @@
<?php
namespace Laravel\Reverb\Concerns;
use Laravel\Reverb\Contracts\ApplicationProvider;
trait SerializesConnections
{
/**
* Prepare the connection instance values for serialization.
*
* @return array<string, mixed>
*/
public function __serialize(): array
{
return [
'id' => $this->id(),
'identifier' => $this->identifier(),
'application' => $this->app()->id(),
'origin' => $this->origin(),
'lastSeenAt' => $this->lastSeenAt,
'hasBeenPinged' => $this->hasBeenPinged,
];
}
/**
* Restore the connection after serialization.
*/
public function __unserialize(array $values): void
{
$this->id = $values['id'];
$this->identifier = $values['identifier'];
$this->application = app(ApplicationProvider::class)->findById($values['application']);
$this->origin = $values['origin'];
$this->lastSeenAt = $values['lastSeenAt'] ?? null;
$this->hasBeenPinged = $values['hasBeenPinged'] ?? null;
}
}
+78
View File
@@ -0,0 +1,78 @@
<?php
namespace Laravel\Reverb;
use Illuminate\Support\Collection;
use Laravel\Reverb\Contracts\ApplicationProvider;
use Laravel\Reverb\Exceptions\InvalidApplication;
class ConfigApplicationProvider implements ApplicationProvider
{
/**
* Create a new config provider instance.
*/
public function __construct(protected Collection $applications)
{
//
}
/**
* Get all of the configured applications as Application instances.
*
* @return \Illuminate\Support\Collection<\Laravel\Reverb\Application>
*/
public function all(): Collection
{
return $this->applications->map(function ($app) {
return $this->findById($app['app_id']);
});
}
/**
* Find an application instance by ID.
*
* @throws \Laravel\Reverb\Exceptions\InvalidApplication
*/
public function findById(string $id): Application
{
return $this->find('app_id', $id);
}
/**
* Find an application instance by key.
*
* @throws \Laravel\Reverb\Exceptions\InvalidApplication
*/
public function findByKey(string $key): Application
{
return $this->find('key', $key);
}
/**
* Find an application instance.
*
* @throws \Laravel\Reverb\Exceptions\InvalidApplication
*/
public function find(string $key, mixed $value): Application
{
$app = $this->applications->firstWhere($key, $value);
if (! $app) {
throw new InvalidApplication;
}
return new Application(
$app['app_id'],
$app['key'],
$app['secret'],
$app['ping_interval'],
$app['activity_timeout'] ?? 30,
$app['allowed_origins'],
$app['max_message_size'],
$app['max_connections'] ?? null,
// If no setting is provided, default to allowing all client events...
$app['accept_client_events_from'] ?? 'all',
$app['options'] ?? [],
);
}
}
+69
View File
@@ -0,0 +1,69 @@
<?php
namespace Laravel\Reverb;
use Laravel\Reverb\Concerns\GeneratesIdentifiers;
use Laravel\Reverb\Contracts\Connection as ConnectionContract;
use Laravel\Reverb\Events\MessageSent;
use Ratchet\RFC6455\Messaging\Frame;
class Connection extends ConnectionContract
{
use GeneratesIdentifiers;
/**
* The normalized socket ID.
*/
protected ?string $id = null;
/**
* Stores the ping state of the connection.
*/
protected $hasBeenPinged = false;
/**
* Get the raw socket connection identifier.
*/
public function identifier(): string
{
return (string) $this->connection->id();
}
/**
* Get the normalized socket ID.
*/
public function id(): string
{
if (! $this->id) {
$this->id = $this->generateId();
}
return $this->id;
}
/**
* Send a message to the connection.
*/
public function send(string $message): void
{
$this->connection->send($message);
MessageSent::dispatch($this, $message);
}
/**
* Send a control frame to the connection.
*/
public function control(string $type = Frame::OP_PING): void
{
$this->connection->send(new Frame('', opcode: $type));
}
/**
* Terminate a connection.
*/
public function terminate(): void
{
$this->connection->close();
}
}
@@ -0,0 +1,188 @@
<?php
namespace Laravel\Reverb\Console\Commands;
use Illuminate\Console\Command;
use Illuminate\Support\Arr;
use Illuminate\Support\Facades\File;
use Illuminate\Support\Str;
use Symfony\Component\Console\Attribute\AsCommand;
use function Laravel\Prompts\confirm;
#[AsCommand(name: 'reverb:install')]
class InstallCommand extends Command
{
/**
* The name and signature of the console command.
*
* @var string
*/
protected $signature = 'reverb:install';
/**
* The console command description.
*
* @var string
*/
protected $description = 'Install the Reverb dependencies';
/**
* Execute the console command.
*/
public function handle(): void
{
$this->addEnvironmentVariables();
$this->publishConfiguration();
$this->updateBroadcastingConfiguration();
$this->enableBroadcasting();
$this->updateBroadcastingDriver();
$this->components->info('Reverb installed successfully.');
}
/**
* Add the Reverb variables to the environment file.
*/
protected function addEnvironmentVariables(): void
{
if (File::missing($env = app()->environmentFile())) {
return;
}
$contents = File::get($env);
$appId = random_int(100_000, 999_999);
$appKey = Str::lower(Str::random(20));
$appSecret = Str::lower(Str::random(20));
$variables = Arr::where([
'REVERB_APP_ID' => "REVERB_APP_ID={$appId}",
'REVERB_APP_KEY' => "REVERB_APP_KEY={$appKey}",
'REVERB_APP_SECRET' => "REVERB_APP_SECRET={$appSecret}",
'REVERB_HOST' => 'REVERB_HOST="localhost"',
'REVERB_PORT' => 'REVERB_PORT=8080',
'REVERB_SCHEME' => 'REVERB_SCHEME=http',
'REVERB_NEW_LINE' => null,
'VITE_REVERB_APP_KEY' => 'VITE_REVERB_APP_KEY="${REVERB_APP_KEY}"',
'VITE_REVERB_HOST' => 'VITE_REVERB_HOST="${REVERB_HOST}"',
'VITE_REVERB_PORT' => 'VITE_REVERB_PORT="${REVERB_PORT}"',
'VITE_REVERB_SCHEME' => 'VITE_REVERB_SCHEME="${REVERB_SCHEME}"',
], function ($value, $key) use ($contents) {
return ! Str::contains($contents, PHP_EOL.$key);
});
$variables = trim(implode(PHP_EOL, $variables));
if ($variables === '') {
return;
}
File::append(
$env,
Str::endsWith($contents, PHP_EOL) ? PHP_EOL.$variables.PHP_EOL : PHP_EOL.PHP_EOL.$variables.PHP_EOL,
);
}
/**
* Publish the Reverb configuration file.
*/
protected function publishConfiguration(): void
{
$this->callSilently('vendor:publish', [
'--provider' => 'Laravel\Reverb\ReverbServiceProvider',
'--tag' => 'reverb-config',
]);
}
/**
* Update the broadcasting.php configuration file.
*/
protected function updateBroadcastingConfiguration(): void
{
if ($this->laravel->config->has('broadcasting.connections.reverb')) {
return;
}
File::replaceInFile(
"'connections' => [\n",
<<<'CONFIG'
'connections' => [
'reverb' => [
'driver' => 'reverb',
'key' => env('REVERB_APP_KEY'),
'secret' => env('REVERB_APP_SECRET'),
'app_id' => env('REVERB_APP_ID'),
'options' => [
'host' => env('REVERB_HOST'),
'port' => env('REVERB_PORT', 443),
'scheme' => env('REVERB_SCHEME', 'https'),
'useTLS' => env('REVERB_SCHEME', 'https') === 'https',
],
'client_options' => [
// Guzzle client options: https://docs.guzzlephp.org/en/stable/request-options.html
],
],
CONFIG,
app()->configPath('broadcasting.php')
);
}
/**
* Enable Laravel's broadcasting functionality.
*/
protected function enableBroadcasting(): void
{
$this->enableBroadcastServiceProvider();
if (File::exists(base_path('routes/channels.php'))) {
return;
}
$enable = confirm('Would you like to enable event broadcasting?', default: true);
if (! $enable) {
return;
}
if ($this->getApplication()->has('install:broadcasting')) {
$this->call('install:broadcasting', ['--no-interaction' => true]);
}
}
/**
* Uncomment the "BroadcastServiceProvider" in the application configuration.
*/
protected function enableBroadcastServiceProvider(): void
{
$config = File::get(app()->configPath('app.php'));
if (Str::contains($config, '// App\Providers\BroadcastServiceProvider::class')) {
File::replaceInFile(
'// App\Providers\BroadcastServiceProvider::class',
'App\Providers\BroadcastServiceProvider::class',
app()->configPath('app.php'),
);
}
}
/**
* Update the configured broadcasting driver.
*/
protected function updateBroadcastingDriver(): void
{
$enable = confirm('Would you like to enable the Reverb broadcasting driver?', default: true);
if (! $enable || File::missing($env = app()->environmentFile())) {
return;
}
File::put(
$env,
Str::of(File::get($env))->replaceMatches('/(BROADCAST_(?:DRIVER|CONNECTION))=.*/', function (array $matches) {
return $matches[1].'=reverb';
})
);
}
}
@@ -0,0 +1,39 @@
<?php
namespace Laravel\Reverb\Console\Components;
use Illuminate\Console\View\Components\Component;
use Symfony\Component\Console\Output\OutputInterface;
class Message extends Component
{
/**
* Renders the component using the given arguments.
*/
public function render(string $message, int $verbosity = OutputInterface::VERBOSITY_NORMAL): void
{
$this->renderView('message', [
'message' => $message,
], $verbosity);
}
/**
* Compile the given view contents.
*
* @param string $view
* @param array $data
* @return void
*/
protected function compile($view, $data)
{
extract($data);
ob_start();
include __DIR__."/views/$view.php";
return tap(ob_get_contents(), function () {
ob_end_clean();
});
}
}
@@ -0,0 +1,5 @@
<div class="flex mx-1 mb-1">
<code>
<?php echo htmlspecialchars($message) ?>
</code>
</div>
@@ -0,0 +1,30 @@
<?php
namespace Laravel\Reverb\Contracts;
use Illuminate\Support\Collection;
use Laravel\Reverb\Application;
interface ApplicationProvider
{
/**
* Get all of the configured applications as Application instances.
*
* @return \Illuminate\Support\Collection<\Laravel\Reverb\Application>
*/
public function all(): Collection;
/**
* Find an application instance by ID.
*
* @throws \Laravel\Reverb\Exceptions\InvalidApplication
*/
public function findById(string $id): Application;
/**
* Find an application instance by key.
*
* @throws \Laravel\Reverb\Exceptions\InvalidApplication
*/
public function findByKey(string $key): Application;
}
+168
View File
@@ -0,0 +1,168 @@
<?php
namespace Laravel\Reverb\Contracts;
use Laravel\Reverb\Application;
use Ratchet\RFC6455\Messaging\Frame;
abstract class Connection
{
/**
* The last time the connection was seen.
*/
protected ?int $lastSeenAt;
/**
* Stores the ping state of the connection.
*/
protected $hasBeenPinged = false;
/**
* Indicates if the connection uses control frames.
*/
protected $usesControlFrames = false;
/**
* Create a new connection instance.
*/
public function __construct(protected WebSocketConnection $connection, protected Application $application, protected ?string $origin)
{
$this->lastSeenAt = time();
}
/**
* Get the raw socket connection identifier.
*/
abstract public function identifier(): string;
/**
* Get the normalized socket ID.
*/
abstract public function id(): string;
/**
* Send a message to the connection.
*/
abstract public function send(string $message): void;
/**
* Send a control frame to the connection.
*/
abstract public function control(string $type = Frame::OP_PING): void;
/**
* Terminate a connection.
*/
abstract public function terminate(): void;
/**
* Get the application the connection belongs to.
*/
public function app(): Application
{
return $this->application;
}
/**
* Get the origin of the connection.
*/
public function origin(): ?string
{
return $this->origin;
}
/**
* Mark the connection as pinged.
*/
public function ping(): void
{
$this->hasBeenPinged = true;
}
/**
* Mark the connection as ponged.
*/
public function pong(): void
{
$this->hasBeenPinged = false;
}
/**
* Get the last time the connection was seen.
*/
public function lastSeenAt(): ?int
{
return $this->lastSeenAt;
}
/**
* Set the connection last seen at timestamp.
*/
public function setLastSeenAt(int $time): Connection
{
$this->lastSeenAt = $time;
return $this;
}
/**
* Touch the connection last seen at timestamp.
*/
public function touch(): Connection
{
$this->setLastSeenAt(time());
$this->pong();
return $this;
}
/**
* Disconnect and unsubscribe from all channels.
*/
public function disconnect(): void
{
$this->terminate();
}
/**
* Determine whether the connection is still active.
*/
public function isActive(): bool
{
return time() < $this->lastSeenAt + $this->app()->pingInterval();
}
/**
* Determine whether the connection is inactive.
*/
public function isInactive(): bool
{
return ! $this->isActive();
}
/**
* Determine whether the connection is stale.
*/
public function isStale(): bool
{
return $this->isInactive() && $this->hasBeenPinged;
}
/**
* Determine whether the connection uses control frames.
*/
public function usesControlFrames(): bool
{
return $this->usesControlFrames;
}
/**
* Mark the connection as using control frames to track activity.
*/
public function setUsesControlFrames(bool $usesControlFrames = true): Connection
{
$this->usesControlFrames = $usesControlFrames;
return $this;
}
}
+26
View File
@@ -0,0 +1,26 @@
<?php
namespace Laravel\Reverb\Contracts;
interface Logger
{
/**
* Log an informational message.
*/
public function info(string $title, ?string $message = null): void;
/**
* Log an error message.
*/
public function error(string $message): void;
/**
* Log a message sent to the server.
*/
public function message(string $message): void;
/**
* Append a new line to the log.
*/
public function line(int $lines = 1): void;
}
+54
View File
@@ -0,0 +1,54 @@
<?php
namespace Laravel\Reverb\Contracts;
abstract class ServerProvider
{
/**
* Bootstrap any application services.
*/
public function boot(): void
{
//
}
/**
* Register any application services.
*/
public function register(): void
{
//
}
/**
* Determine whether the server should publish events.
*/
public function shouldPublishEvents(): bool
{
return false;
}
/**
* Determine whether the server subscribes to events.
*/
public function subscribesToEvents(): bool
{
return $this->shouldPublishEvents();
}
/**
* Determine whether the server should not publish events.
*/
public function shouldNotPublishEvents(): bool
{
return ! $this->shouldPublishEvents();
}
/**
* Determine whether the server should not subscribe to events.
*/
public function doesNotSubscribeToEvents(): bool
{
return ! $this->subscribesToEvents();
}
}
@@ -0,0 +1,21 @@
<?php
namespace Laravel\Reverb\Contracts;
interface WebSocketConnection
{
/**
* Get the raw socket connection identifier.
*/
public function id(): int|string;
/**
* Send a message to the connection.
*/
public function send(mixed $message): void;
/**
* Close the connection.
*/
public function close(mixed $message = null): void;
}
+19
View File
@@ -0,0 +1,19 @@
<?php
namespace Laravel\Reverb\Events;
use Illuminate\Foundation\Events\Dispatchable;
use Laravel\Reverb\Protocols\Pusher\Channels\Channel;
class ChannelCreated
{
use Dispatchable;
/**
* Create a new event instance.
*/
public function __construct(public Channel $channel)
{
//
}
}
+19
View File
@@ -0,0 +1,19 @@
<?php
namespace Laravel\Reverb\Events;
use Illuminate\Foundation\Events\Dispatchable;
use Laravel\Reverb\Protocols\Pusher\Channels\Channel;
class ChannelRemoved
{
use Dispatchable;
/**
* Create a new event instance.
*/
public function __construct(public Channel $channel)
{
//
}
}
+19
View File
@@ -0,0 +1,19 @@
<?php
namespace Laravel\Reverb\Events;
use Illuminate\Foundation\Events\Dispatchable;
use Laravel\Reverb\Protocols\Pusher\Channels\ChannelConnection;
class ConnectionPruned
{
use Dispatchable;
/**
* Create a new event instance.
*/
public function __construct(public ChannelConnection $connection)
{
//
}
}
+19
View File
@@ -0,0 +1,19 @@
<?php
namespace Laravel\Reverb\Events;
use Illuminate\Foundation\Events\Dispatchable;
use Laravel\Reverb\Contracts\Connection;
class MessageReceived
{
use Dispatchable;
/**
* Create a new event instance.
*/
public function __construct(public Connection $connection, public string $message)
{
//
}
}
+19
View File
@@ -0,0 +1,19 @@
<?php
namespace Laravel\Reverb\Events;
use Illuminate\Foundation\Events\Dispatchable;
use Laravel\Reverb\Contracts\Connection;
class MessageSent
{
use Dispatchable;
/**
* Create a new event instance.
*/
public function __construct(public Connection $connection, public string $message)
{
//
}
}
@@ -0,0 +1,15 @@
<?php
namespace Laravel\Reverb\Exceptions;
use Exception;
class InvalidApplication extends Exception
{
/**
* The error message associated with the exception.
*
* @var string
*/
protected $message = 'Application does not exist';
}
+15
View File
@@ -0,0 +1,15 @@
<?php
namespace Laravel\Reverb\Exceptions;
use Exception;
class InvalidOrigin extends Exception
{
/**
* The error message associated with the exception.
*
* @var string
*/
protected $message = 'Origin not allowed';
}
@@ -0,0 +1,16 @@
<?php
namespace Laravel\Reverb\Exceptions;
use Exception;
class RedisConnectionException extends Exception
{
/**
* Timeout while attempting to connect to Redis.
*/
public static function failedAfter(string $name, int $timeout): self
{
return new static("Failed to connect to Redis connection [{$name}] after retrying for {$timeout}s.");
}
}
@@ -0,0 +1,38 @@
<?php
namespace Laravel\Reverb\Jobs;
use Illuminate\Foundation\Bus\Dispatchable;
use Laravel\Reverb\Contracts\ApplicationProvider;
use Laravel\Reverb\Loggers\Log;
use Laravel\Reverb\Protocols\Pusher\Contracts\ChannelManager;
use Laravel\Reverb\Protocols\Pusher\EventHandler;
class PingInactiveConnections
{
use Dispatchable;
/**
* Execute the job.
*/
public function handle(ChannelManager $channels): void
{
Log::info('Pinging Inactive Connections');
$pusher = new EventHandler($channels);
app(ApplicationProvider::class)
->all()
->each(function ($application) use ($channels, $pusher) {
foreach ($channels->for($application)->connections() as $connection) {
if ($connection->isActive()) {
continue;
}
$pusher->ping($connection->connection());
Log::info('Connection Pinged', $connection->id());
}
});
}
}
@@ -0,0 +1,50 @@
<?php
namespace Laravel\Reverb\Jobs;
use Illuminate\Foundation\Bus\Dispatchable;
use Laravel\Reverb\Contracts\ApplicationProvider;
use Laravel\Reverb\Events\ConnectionPruned;
use Laravel\Reverb\Loggers\Log;
use Laravel\Reverb\Protocols\Pusher\Contracts\ChannelManager;
class PruneStaleConnections
{
use Dispatchable;
/**
* Execute the job.
*/
public function handle(ChannelManager $channels): void
{
Log::info('Pruning Stale Connections');
app(ApplicationProvider::class)
->all()
->each(function ($application) use ($channels) {
foreach ($channels->for($application)->connections() as $connection) {
if (! $connection->isStale()) {
continue;
}
$connection->send(json_encode([
'event' => 'pusher:error',
'data' => json_encode([
'code' => 4201,
'message' => 'Pong reply not received in time',
]),
]));
$channels
->for($connection->app())
->unsubscribeFromAll($connection->connection());
$connection->disconnect();
Log::info('Connection Pruned', $connection->id());
ConnectionPruned::dispatch($connection);
}
});
}
}
+73
View File
@@ -0,0 +1,73 @@
<?php
namespace Laravel\Reverb\Loggers;
use Illuminate\Console\OutputStyle;
use Illuminate\Console\View\Components\Factory;
use Illuminate\Support\Str;
use Laravel\Reverb\Console\Components\Message;
use Laravel\Reverb\Contracts\Logger;
class CliLogger implements Logger
{
/**
* The components factory instance.
*
* @var \Illuminate\Console\View\Components\Factory
*/
protected $components;
/**
* Create a new CLI logger instance.
*/
public function __construct(protected OutputStyle $output)
{
$this->components = new Factory($output);
}
/**
* Log an informational message.
*/
public function info(string $title, ?string $message = null): void
{
$this->components->twoColumnDetail($title, $message);
}
/**
* Log an error message.
*/
public function error(string $string): void
{
$this->output->error($string);
}
/**
* Log a message sent to the server.
*/
public function message(string $message): void
{
$message = json_decode($message, true);
if (isset($message['data']) && is_string($message['data'])) {
$message['data'] = json_decode($message['data'], true);
}
if (isset($message['data']['channel_data']) && is_string($message['data']['channel_data'])) {
$message['data']['channel_data'] = json_decode($message['data']['channel_data'], true);
}
$message = json_encode($message, JSON_PRETTY_PRINT);
(new Message($this->output))->render(
Str::limit($message, 200)
);
}
/**
* Append a new line to the log.
*/
public function line(int $lines = 1): void
{
$this->output->newLine($lines);
}
}
+29
View File
@@ -0,0 +1,29 @@
<?php
namespace Laravel\Reverb\Loggers;
use Laravel\Reverb\Contracts\Logger;
class Log
{
/**
* The logger instance.
*
* @var \Laravel\Reverb\Contracts\Logger
*/
protected static $logger;
/**
* Proxy method calls to the logger instance.
*
* @param string $method
* @param array $arguments
* @return mixed
*/
public static function __callStatic($method, $arguments)
{
static::$logger ??= app(Logger::class);
return static::$logger->{$method}(...$arguments);
}
}
+40
View File
@@ -0,0 +1,40 @@
<?php
namespace Laravel\Reverb\Loggers;
use Laravel\Reverb\Contracts\Logger;
class NullLogger implements Logger
{
/**
* Log an informational message.
*/
public function info(string $title, ?string $message = null): void
{
//
}
/**
* Log an error message.
*/
public function error(string $string): void
{
//
}
/**
* Log a message sent to the server.
*/
public function message(string $message): void
{
//
}
/**
* Append a new line to the log.
*/
public function line(int $lines = 1): void
{
//
}
}
+55
View File
@@ -0,0 +1,55 @@
<?php
namespace Laravel\Reverb\Loggers;
use Illuminate\Support\Facades\Log;
use Laravel\Reverb\Contracts\Logger;
class StandardLogger implements Logger
{
/**
* Log an informational message
*/
public function info(string $title, ?string $message = null): void
{
$output = $title;
if ($message) {
$output .= ': '.$message;
}
Log::info($output);
}
/**
* Log an error message.
*/
public function error(string $string): void
{
Log::error($string);
}
/**
* Log a message sent to the server.
*/
public function message(string $message): void
{
$message = json_decode($message, true);
if (isset($message['data']['channel_data'])) {
$message['data']['channel_data'] = json_decode($message['data']['channel_data'], true);
}
$message = json_encode($message, JSON_PRETTY_PRINT);
Log::info($message);
}
/**
* Append a new line to the log.
*/
public function line(int $lines = 1): void
{
//
}
}
@@ -0,0 +1,47 @@
<?php
namespace Laravel\Reverb\Protocols\Pusher\Channels;
use Laravel\Reverb\Contracts\Connection;
class CacheChannel extends Channel
{
/**
* Data from last event triggered.
*/
protected ?array $payload = null;
/**
* Send a message to all connections subscribed to the channel.
*/
public function broadcast(array $payload, ?Connection $except = null): void
{
$this->payload = $payload;
parent::broadcast($payload, $except);
}
/**
* Broadcast a message triggered from an internal source.
*/
public function broadcastInternally(array $payload, ?Connection $except = null): void
{
parent::broadcast($payload, $except);
}
/**
* Determine if the channel has a cached payload.
*/
public function hasCachedPayload(): bool
{
return $this->payload !== null;
}
/**
* Get the cached payload.
*/
public function cachedPayload(): ?array
{
return $this->payload;
}
}
@@ -0,0 +1,147 @@
<?php
namespace Laravel\Reverb\Protocols\Pusher\Channels;
use Laravel\Reverb\Contracts\Connection;
use Laravel\Reverb\Loggers\Log;
use Laravel\Reverb\Protocols\Pusher\Concerns\SerializesChannels;
use Laravel\Reverb\Protocols\Pusher\Contracts\ChannelConnectionManager;
use Laravel\Reverb\Protocols\Pusher\Contracts\ChannelManager;
class Channel
{
use SerializesChannels;
/**
* The channel connections.
*
* @var \Laravel\Reverb\Contracts\ChannelConnectionManager
*/
protected $connections;
/**
* Create a new channel instance.
*/
public function __construct(protected string $name)
{
$this->connections = app(ChannelConnectionManager::class)->for($this->name);
}
/**
* Get the channel name.
*/
public function name(): string
{
return $this->name;
}
/**
* Get all connections for the channel.
*
* @return array<string, \Laravel\Reverb\Protocols\Pusher\Channels\ChannelConnection>
*/
public function connections(): array
{
return $this->connections->all();
}
/**
* Find a connection.
*/
public function find(Connection $connection): ?ChannelConnection
{
return $this->connections->find($connection);
}
/**
* Find a connection by its ID.
*/
public function findById(string $id): ?ChannelConnection
{
return $this->connections->findById($id);
}
/**
* Subscribe to the given channel.
*/
public function subscribe(Connection $connection, ?string $auth = null, ?string $data = null): void
{
$this->connections->add($connection, $data ? json_decode($data, associative: true, flags: JSON_THROW_ON_ERROR) : []);
}
/**
* Unsubscribe from the given channel.
*/
public function unsubscribe(Connection $connection): void
{
$this->connections->remove($connection);
if ($this->connections->isEmpty()) {
app(ChannelManager::class)->for($connection->app())->remove($this);
}
}
/**
* Determine if the connection is subscribed to the channel.
*/
public function subscribed(Connection $connection): bool
{
return $this->connections->find($connection) !== null;
}
/**
* Send a message to all connections subscribed to the channel.
*/
public function broadcast(array $payload, ?Connection $except = null): void
{
if ($except === null) {
$this->broadcastToAll($payload);
return;
}
$message = json_encode($payload);
Log::info('Broadcasting To', $this->name());
Log::message($message);
foreach ($this->connections() as $connection) {
if ($except->id() === $connection->id()) {
continue;
}
$connection->send($message);
}
}
/**
* Send a broadcast to all connections.
*/
public function broadcastToAll(array $payload): void
{
$message = json_encode($payload);
Log::info('Broadcasting To', $this->name());
Log::message($message);
foreach ($this->connections() as $connection) {
$connection->send($message);
}
}
/**
* Broadcast a message triggered from an internal source.
*/
public function broadcastInternally(array $payload, ?Connection $except = null): void
{
$this->broadcast($payload, $except);
}
/**
* Get the data associated with the channel.
*/
public function data(): array
{
return [];
}
}
@@ -0,0 +1,23 @@
<?php
namespace Laravel\Reverb\Protocols\Pusher\Channels;
use Illuminate\Support\Str;
class ChannelBroker
{
/**
* Return the relevant channel instance.
*/
public static function create(string $name): Channel
{
return match (true) {
Str::startsWith($name, 'private-cache-') => new PrivateCacheChannel($name),
Str::startsWith($name, 'presence-cache-') => new PresenceCacheChannel($name),
Str::startsWith($name, 'cache') => new CacheChannel($name),
Str::startsWith($name, 'private') => new PrivateChannel($name),
Str::startsWith($name, 'presence') => new PresenceChannel($name),
default => new Channel($name),
};
}
}
@@ -0,0 +1,49 @@
<?php
namespace Laravel\Reverb\Protocols\Pusher\Channels;
use Illuminate\Support\Arr;
use Laravel\Reverb\Contracts\Connection;
class ChannelConnection
{
/**
* Create a new channel connection instance.
*/
public function __construct(protected Connection $connection, protected array $data = [])
{
//
}
/**
* Get the underlying connection.
*/
public function connection(): Connection
{
return $this->connection;
}
/**
* Get the connection data.
*/
public function data(?string $key = null): mixed
{
return $key ? Arr::get($this->data, $key) : $this->data;
}
/**
* Send a message to the connection.
*/
public function send(string $message): void
{
$this->connection->send($message);
}
/**
* Proxy the given method to the underlying connection.
*/
public function __call(string $method, array $parameters): mixed
{
return $this->connection->{$method}(...$parameters);
}
}
@@ -0,0 +1,104 @@
<?php
namespace Laravel\Reverb\Protocols\Pusher\Channels\Concerns;
use Laravel\Reverb\Contracts\Connection;
trait InteractsWithPresenceChannels
{
use InteractsWithPrivateChannels;
/**
* Subscribe to the given channel.
*/
public function subscribe(Connection $connection, ?string $auth = null, ?string $data = null): void
{
$this->verify($connection, $auth, $data);
$userData = $data ? json_decode($data, associative: true, flags: JSON_THROW_ON_ERROR) : [];
if ($this->userIsSubscribed($userData['user_id'] ?? null)) {
parent::subscribe($connection, $auth, $data);
return;
}
parent::subscribe($connection, $auth, $data);
parent::broadcastInternally(
[
'event' => 'pusher_internal:member_added',
'data' => json_encode((object) $userData),
'channel' => $this->name(),
],
$connection
);
}
/**
* Unsubscribe from the given channel.
*/
public function unsubscribe(Connection $connection): void
{
$subscription = $this->connections->find($connection);
parent::unsubscribe($connection);
if (
! $subscription ||
! $subscription->data('user_id') ||
$this->userIsSubscribed($subscription->data('user_id'))
) {
return;
}
parent::broadcast(
[
'event' => 'pusher_internal:member_removed',
'data' => json_encode(['user_id' => $subscription->data('user_id')]),
'channel' => $this->name(),
],
$connection
);
}
/**
* Get the data associated with the channel.
*/
public function data(): array
{
$connections = collect($this->connections->all())
->map(fn ($connection) => $connection->data())
->unique('user_id');
if ($connections->contains(fn ($connection) => ! isset($connection['user_id']))) {
return [
'presence' => [
'count' => 0,
'ids' => [],
'hash' => [],
],
];
}
return [
'presence' => [
'count' => $connections->count() ?? 0,
'ids' => $connections->map(fn ($connection) => $connection['user_id'])->values()->all(),
'hash' => $connections->keyBy('user_id')->map->user_info->toArray(),
],
];
}
/**
* Determine if the given user is subscribed to the channel.
*/
protected function userIsSubscribed(?string $userId): bool
{
if (! $userId) {
return false;
}
return collect($this->connections->all())->map(fn ($connection) => (string) $connection->data('user_id'))->contains($userId);
}
}
@@ -0,0 +1,45 @@
<?php
namespace Laravel\Reverb\Protocols\Pusher\Channels\Concerns;
use Illuminate\Support\Str;
use Laravel\Reverb\Contracts\Connection;
use Laravel\Reverb\Protocols\Pusher\Exceptions\ConnectionUnauthorized;
trait InteractsWithPrivateChannels
{
/**
* Subscribe to the given channel.
*/
public function subscribe(Connection $connection, ?string $auth = null, ?string $data = null): void
{
$this->verify($connection, $auth, $data);
parent::subscribe($connection, $auth, $data);
}
/**
* Determine whether the given authentication token is valid.
*/
protected function verify(Connection $connection, ?string $auth = null, ?string $data = null): bool
{
$signature = "{$connection->id()}:{$this->name()}";
if ($data) {
$signature .= ":{$data}";
}
if (! hash_equals(
hash_hmac(
'sha256',
$signature,
$connection->app()->secret(),
),
Str::after($auth, ':')
)) {
throw new ConnectionUnauthorized;
}
return true;
}
}
@@ -0,0 +1,10 @@
<?php
namespace Laravel\Reverb\Protocols\Pusher\Channels;
use Laravel\Reverb\Protocols\Pusher\Channels\Concerns\InteractsWithPresenceChannels;
class PresenceCacheChannel extends CacheChannel
{
use InteractsWithPresenceChannels;
}
@@ -0,0 +1,10 @@
<?php
namespace Laravel\Reverb\Protocols\Pusher\Channels;
use Laravel\Reverb\Protocols\Pusher\Channels\Concerns\InteractsWithPresenceChannels;
class PresenceChannel extends PrivateChannel
{
use InteractsWithPresenceChannels;
}
@@ -0,0 +1,10 @@
<?php
namespace Laravel\Reverb\Protocols\Pusher\Channels;
use Laravel\Reverb\Protocols\Pusher\Channels\Concerns\InteractsWithPrivateChannels;
class PrivateCacheChannel extends CacheChannel
{
use InteractsWithPrivateChannels;
}
@@ -0,0 +1,10 @@
<?php
namespace Laravel\Reverb\Protocols\Pusher\Channels;
use Laravel\Reverb\Protocols\Pusher\Channels\Concerns\InteractsWithPrivateChannels;
class PrivateChannel extends Channel
{
use InteractsWithPrivateChannels;
}
@@ -0,0 +1,95 @@
<?php
namespace Laravel\Reverb\Protocols\Pusher;
use Illuminate\Support\Facades\Validator;
use Illuminate\Support\Str;
use Laravel\Reverb\Contracts\Connection;
use Laravel\Reverb\Protocols\Pusher\Contracts\ChannelManager;
class ClientEvent
{
/**
* Handle a Pusher client event.
*/
public static function handle(Connection $connection, array $event): void
{
Validator::make($event, [
'event' => ['required', 'string'],
'channel' => ['required', 'string'],
'data' => ['nullable', 'array'],
])->validate();
if (! Str::startsWith($event['event'], 'client-')) {
return;
}
if (! isset($event['channel'])) {
return;
}
$acceptClientEventsFrom = $connection->app()->acceptClientEventsFrom();
if (! in_array($acceptClientEventsFrom, ['all', 'members'])) {
// Client events are disabled, so we should reject the event...
$connection->send(json_encode([
'event' => 'pusher:error',
'data' => json_encode([
'code' => 4301,
'message' => 'The app does not have client messaging enabled.',
]),
]));
return;
}
$rebroadcastEvent = $event;
if ($acceptClientEventsFrom == 'members') {
$channel = app(ChannelManager::class)->find($event['channel']);
$channelConnection = $channel?->find($connection);
if (! $channelConnection) {
$connection->send(json_encode([
'event' => 'pusher:error',
'data' => json_encode([
'code' => 4009,
'message' => 'The client is not a member of the specified channel.',
]),
]));
return;
}
// Regenerate event payload, ensuring we only include the expected fields and the authenticated user_id...
$rebroadcastEvent = [
'event' => $event['event'],
'channel' => $event['channel'],
'data' => $event['data'] ?? null,
];
if ($userId = $channelConnection->data('user_id')) {
// Because public channels allow unauthenticated users, we may not have a user ID...
$rebroadcastEvent['user_id'] = $userId;
}
}
self::whisper(
$connection,
$rebroadcastEvent
);
}
/**
* Whisper a message to all connections on the channel associated with the event.
*/
public static function whisper(Connection $connection, array $payload): void
{
EventDispatcher::dispatch(
$connection->app(),
$payload,
$connection
);
}
}
@@ -0,0 +1,93 @@
<?php
namespace Laravel\Reverb\Protocols\Pusher\Concerns;
use Laravel\Reverb\Application;
use Laravel\Reverb\Protocols\Pusher\Channels\CacheChannel;
use Laravel\Reverb\Protocols\Pusher\Channels\Channel;
use Laravel\Reverb\Protocols\Pusher\Channels\Concerns\InteractsWithPresenceChannels;
use Laravel\Reverb\Protocols\Pusher\Contracts\ChannelManager;
trait InteractsWithChannelInformation
{
/**
* Get meta / status information for the given channels.
*/
protected function infoForChannels(Application $application, array $channels, string $info): array
{
return collect($channels)->mapWithKeys(function ($channel) use ($application, $info) {
$name = $channel instanceof Channel ? $channel->name() : $channel;
return [$name => $this->info($application, $name, $info)];
})->all();
}
/**
* Get meta / status information for the given channel.
*
* @return array<string, array<string, int>>
*/
protected function info(Application $application, string $channel, string $info): array
{
$info = explode(',', $info);
$channel = app(ChannelManager::class)->for($application)->find($channel);
return array_filter(
$channel ? $this->occupiedInfo($channel, $info) : $this->unoccupiedInfo($info),
fn ($item) => $item !== null
);
}
/**
* Get channel information for the given occupied channel.
*/
private function occupiedInfo(Channel $channel, array $info): array
{
$count = count($channel->connections());
return [
'occupied' => in_array('occupied', $info) ? $count > 0 : null,
'user_count' => in_array('user_count', $info) && $this->isPresenceChannel($channel) ? $this->userCount($channel) : null,
'subscription_count' => in_array('subscription_count', $info) && ! $this->isPresenceChannel($channel) ? $count : null,
'cache' => in_array('cache', $info) && $this->isCacheChannel($channel) ? $channel->cachedPayload() : null,
];
}
/**
* Get channel information for the given unoccupied channel.
*/
private function unoccupiedInfo(array $info): array
{
return [
'occupied' => in_array('occupied', $info) ? false : null,
];
}
/**
* Determine if the given channel is a presence channel.
*/
protected function isPresenceChannel(Channel $channel): bool
{
return in_array(InteractsWithPresenceChannels::class, class_uses($channel));
}
/**
* Determine if the given channel is a cache channel.
*/
protected function isCacheChannel(Channel $channel): bool
{
return $channel instanceof CacheChannel;
}
/**
* Get the number of unique users subscribed to the presence channel.
*/
protected function userCount(Channel $channel): int
{
return collect($channel->connections())
->map(fn ($connection) => $connection->data())
->unique('user_id')
->count();
}
}
@@ -0,0 +1,29 @@
<?php
namespace Laravel\Reverb\Protocols\Pusher\Concerns;
use Laravel\Reverb\Protocols\Pusher\Contracts\ChannelConnectionManager;
trait SerializesChannels
{
/**
* Prepare the channel instance values for serialization.
*
* @return array<string, mixed>
*/
public function __serialize(): array
{
return [
'name' => $this->name,
];
}
/**
* Restore the channel after serialization.
*/
public function __unserialize(array $values): void
{
$this->name = $values['name'];
$this->connections = app(ChannelConnectionManager::class)->for($this->name);
}
}
@@ -0,0 +1,51 @@
<?php
namespace Laravel\Reverb\Protocols\Pusher\Contracts;
use Laravel\Reverb\Contracts\Connection;
use Laravel\Reverb\Protocols\Pusher\Channels\ChannelConnection;
interface ChannelConnectionManager
{
/**
* Get a channel connection manager for the given channel name.
*/
public function for(string $name): ChannelConnectionManager;
/**
* Add a connection.
*/
public function add(Connection $connection, array $data): void;
/**
* Remove a connection.
*/
public function remove(Connection $connection): void;
/**
* Find a connection.
*/
public function find(Connection $connection): ?ChannelConnection;
/**
* Find a connection by its ID.
*/
public function findById(string $id): ?ChannelConnection;
/**
* Get all of the connections.
*
* @return array<string, \Laravel\Reverb\Protocols\Pusher\Channels\ChannelConnection>
*/
public function all(): array;
/**
* Determine whether any connections remain on the channel.
*/
public function isEmpty(): bool;
/**
* Flush the channel connection manager.
*/
public function flush(): void;
}
@@ -0,0 +1,64 @@
<?php
namespace Laravel\Reverb\Protocols\Pusher\Contracts;
use Laravel\Reverb\Application;
use Laravel\Reverb\Contracts\Connection;
use Laravel\Reverb\Protocols\Pusher\Channels\Channel;
interface ChannelManager
{
/**
* Get the application instance.
*/
public function app(): ?Application;
/**
* The application the channel manager should be scoped to.
*/
public function for(Application $application): ChannelManager;
/**
* Get all the channels.
*
* @return array<string, \Laravel\Reverb\Protocols\Pusher\Channels\Channel>
*/
public function all(): array;
/**
* Determine whether the given channel exists.
*/
public function exists(string $channel): bool;
/**
* Find the given channel.
*/
public function find(string $channel): ?Channel;
/**
* Find the given channel or create it if it doesn't exist.
*/
public function findOrCreate(string $channel): Channel;
/**
* Get all the connections for the given channels.
*
* @return array<string, \Laravel\Reverb\Protocols\Pusher\Channels\ChannelConnection>
*/
public function connections(?string $channel = null): array;
/**
* Unsubscribe from all channels.
*/
public function unsubscribeFromAll(Connection $connection): void;
/**
* Remove the given channel.
*/
public function remove(Channel $channel): void;
/**
* Flush the channel manager repository.
*/
public function flush(): void;
}
@@ -0,0 +1,59 @@
<?php
namespace Laravel\Reverb\Protocols\Pusher;
use Illuminate\Support\Arr;
use Laravel\Reverb\Application;
use Laravel\Reverb\Contracts\Connection;
use Laravel\Reverb\Protocols\Pusher\Contracts\ChannelManager;
use Laravel\Reverb\ServerProviderManager;
use Laravel\Reverb\Servers\Reverb\Contracts\PubSubProvider;
class EventDispatcher
{
/**
* Dispatch a message to a channel.
*/
public static function dispatch(Application $app, array $payload, ?Connection $connection = null): void
{
$server = app(ServerProviderManager::class);
if ($server->shouldNotPublishEvents()) {
static::dispatchSynchronously($app, $payload, $connection);
return;
}
$data = [
'type' => 'message',
'application' => serialize($app),
'payload' => $payload,
];
if ($connection?->id() !== null) {
$data['socket_id'] = $connection?->id();
}
app(PubSubProvider::class)->publish($data);
}
/**
* Notify all connections subscribed to the given channel.
*/
public static function dispatchSynchronously(Application $app, array $payload, ?Connection $connection = null): void
{
$channels = Arr::wrap($payload['channels'] ?? $payload['channel'] ?? []);
foreach ($channels as $channel) {
unset($payload['channels']);
if (! $channel = app(ChannelManager::class)->for($app)->find($channel)) {
continue;
}
$payload['channel'] = $channel->name();
$channel->broadcast($payload, $connection);
}
}
}
@@ -0,0 +1,187 @@
<?php
namespace Laravel\Reverb\Protocols\Pusher;
use Exception;
use Illuminate\Support\Facades\Validator;
use Illuminate\Support\Str;
use Laravel\Reverb\Contracts\Connection;
use Laravel\Reverb\Protocols\Pusher\Channels\CacheChannel;
use Laravel\Reverb\Protocols\Pusher\Channels\Channel;
use Laravel\Reverb\Protocols\Pusher\Contracts\ChannelManager;
class EventHandler
{
/**
* Create a new Pusher event instance.
*/
public function __construct(protected ChannelManager $channels)
{
//
}
/**
* Handle an incoming Pusher event.
*/
public function handle(Connection $connection, string $event, array $payload = []): void
{
$event = Str::after($event, 'pusher:');
match ($event) {
'connection_established' => $this->acknowledge($connection),
'subscribe' => $this->subscribe(
$connection,
$payload['channel'],
$payload['auth'] ?? null,
$payload['channel_data'] ?? null
),
'unsubscribe' => $this->unsubscribe($connection, $payload['channel']),
'ping' => $this->pong($connection),
'pong' => $connection->touch(),
default => throw new Exception('Unknown Pusher event: '.$event),
};
}
/**
* Acknowledge the connection.
*/
public function acknowledge(Connection $connection): void
{
$this->send($connection, 'connection_established', [
'socket_id' => $connection->id(),
'activity_timeout' => $connection->app()->activityTimeout(),
]);
}
/**
* Subscribe to the given channel.
*/
public function subscribe(Connection $connection, string $channel, ?string $auth = null, ?string $data = null): void
{
Validator::make([
'channel' => $channel,
'auth' => $auth,
'channel_data' => $data,
], [
'channel' => ['nullable', 'string'],
'auth' => ['nullable', 'string'],
'channel_data' => ['nullable', 'json'],
])->validate();
$channel = $this->channels
->for($connection->app())
->findOrCreate($channel);
$channel->subscribe($connection, $auth, $data);
$this->afterSubscribe($channel, $connection);
}
/**
* Carry out any actions that should be performed after a subscription.
*/
protected function afterSubscribe(Channel $channel, Connection $connection): void
{
$this->sendInternally($connection, 'subscription_succeeded', $channel->data(), $channel->name());
match (true) {
$channel instanceof CacheChannel => $this->sendCachedPayload($channel, $connection),
default => null,
};
}
/**
* Unsubscribe from the given channel.
*/
public function unsubscribe(Connection $connection, string $channel): void
{
$channel = $this->channels
->for($connection->app())
->find($channel)
?->unsubscribe($connection);
}
/**
* Send the cached payload for the given channel.
*/
protected function sendCachedPayload(CacheChannel $channel, Connection $connection): void
{
if ($channel->hasCachedPayload()) {
$connection->send(
json_encode($channel->cachedPayload())
);
return;
}
$this->send($connection, 'cache_miss', channel: $channel->name());
}
/**
* Respond to a ping on the given connection.
*/
public function pong(Connection $connection): void
{
static::send($connection, 'pong');
}
/**
* Send a ping to the given connection.
*/
public function ping(Connection $connection): void
{
$connection->usesControlFrames()
? $connection->control()
: static::send($connection, 'ping');
$connection->ping();
}
/**
* Send a response to the given connection.
*/
public function send(Connection $connection, string $event, array $data = [], ?string $channel = null): void
{
$connection->send(
static::formatPayload($event, $data, $channel)
);
}
/**
* Send an internal response to the given connection.
*/
public function sendInternally(Connection $connection, string $event, array $data = [], ?string $channel = null): void
{
$connection->send(
static::formatInternalPayload($event, $data, $channel)
);
}
/**
* Format the payload for the given event.
*/
public function formatPayload(string $event, array $data = [], ?string $channel = null, string $prefix = 'pusher:'): string|false
{
return json_encode(
array_filter([
'event' => $prefix.$event,
'data' => empty($data) ? null : json_encode($data),
'channel' => $channel,
])
);
}
/**
* Format the internal payload for the given event.
*/
public function formatInternalPayload(string $event, array $data = [], $channel = null): string|false
{
return json_encode(
array_filter([
'event' => 'pusher_internal:'.$event,
'data' => json_encode((object) $data),
'channel' => $channel,
])
);
}
}
@@ -0,0 +1,20 @@
<?php
namespace Laravel\Reverb\Protocols\Pusher\Exceptions;
class ConnectionLimitExceeded extends PusherException
{
/**
* The error code associated with the exception.
*
* @var int
*/
protected $code = 4004;
/**
* The error message associated with the exception.
*
* @var string
*/
protected $message = 'Application is over connection quota';
}
@@ -0,0 +1,20 @@
<?php
namespace Laravel\Reverb\Protocols\Pusher\Exceptions;
class ConnectionUnauthorized extends PusherException
{
/**
* The error code associated with the exception.
*
* @var int
*/
protected $code = 4009;
/**
* The error message associated with the exception.
*
* @var string
*/
protected $message = 'Connection is unauthorized';
}
@@ -0,0 +1,20 @@
<?php
namespace Laravel\Reverb\Protocols\Pusher\Exceptions;
class InvalidOrigin extends PusherException
{
/**
* The error code associated with the exception.
*
* @var int
*/
protected $code = 4009;
/**
* The error message associated with the exception.
*
* @var string
*/
protected $message = 'Origin not allowed';
}
@@ -0,0 +1,46 @@
<?php
namespace Laravel\Reverb\Protocols\Pusher\Exceptions;
use Exception;
abstract class PusherException extends Exception
{
/**
* The error code associated with the exception.
*
* @var int
*/
protected $code;
/**
* The error message associated with the exception.
*
* @var string
*/
protected $message;
/**
* Get the Pusher formatted error payload.
*/
public function payload(): array
{
return [
'event' => 'pusher:error',
'data' => json_encode([
'code' => $this->code,
'message' => $this->message,
]),
];
}
/**
* Get the encoded Pusher formatted error payload.
*/
public function message()
{
return json_encode(
$this->payload()
);
}
}
@@ -0,0 +1,25 @@
<?php
namespace Laravel\Reverb\Protocols\Pusher\Http\Controllers;
use Laravel\Reverb\Protocols\Pusher\MetricsHandler;
use Laravel\Reverb\Servers\Reverb\Http\Connection;
use Laravel\Reverb\Servers\Reverb\Http\Response;
use Psr\Http\Message\RequestInterface;
use React\Promise\PromiseInterface;
class ChannelController extends Controller
{
/**
* Handle the request.
*/
public function __invoke(RequestInterface $request, Connection $connection, string $appId, string $channel): PromiseInterface
{
$this->verify($request, $connection, $appId);
return app(MetricsHandler::class)->gather($this->application, 'channel', [
'channel' => $channel,
'info' => isset($this->query['info']) ? $this->query['info'].',occupied' : 'occupied',
])->then(fn ($channel) => new Response((object) $channel));
}
}
@@ -0,0 +1,37 @@
<?php
namespace Laravel\Reverb\Protocols\Pusher\Http\Controllers;
use Laravel\Reverb\Protocols\Pusher\Concerns\InteractsWithChannelInformation;
use Laravel\Reverb\Protocols\Pusher\MetricsHandler;
use Laravel\Reverb\Servers\Reverb\Http\Connection;
use Laravel\Reverb\Servers\Reverb\Http\Response;
use Psr\Http\Message\RequestInterface;
use React\Promise\PromiseInterface;
class ChannelUsersController extends Controller
{
use InteractsWithChannelInformation;
/**
* Handle the request.
*/
public function __invoke(RequestInterface $request, Connection $connection, string $channel, string $appId): Response|PromiseInterface
{
$this->verify($request, $connection, $appId);
$channel = $this->channels->find($channel);
if (! $channel) {
return new Response((object) [], 404);
}
if (! $this->isPresenceChannel($channel)) {
return new Response((object) [], 400);
}
return app(MetricsHandler::class)
->gather($this->application, 'channel_users', ['channel' => $channel->name()])
->then(fn ($connections) => new Response(['users' => $connections]));
}
}
@@ -0,0 +1,25 @@
<?php
namespace Laravel\Reverb\Protocols\Pusher\Http\Controllers;
use Laravel\Reverb\Protocols\Pusher\MetricsHandler;
use Laravel\Reverb\Servers\Reverb\Http\Connection;
use Laravel\Reverb\Servers\Reverb\Http\Response;
use Psr\Http\Message\RequestInterface;
use React\Promise\PromiseInterface;
class ChannelsController extends Controller
{
/**
* Handle the request.
*/
public function __invoke(RequestInterface $request, Connection $connection, string $appId): PromiseInterface
{
$this->verify($request, $connection, $appId);
return app(MetricsHandler::class)->gather($this->application, 'channels', [
'filter' => $this->query['filter_by_prefix'] ?? null,
'info' => $this->query['info'] ?? null,
])->then(fn ($channels) => new Response(['channels' => array_map(fn ($item) => (object) $item, $channels)]));
}
}
@@ -0,0 +1,23 @@
<?php
namespace Laravel\Reverb\Protocols\Pusher\Http\Controllers;
use Laravel\Reverb\Protocols\Pusher\MetricsHandler;
use Laravel\Reverb\Servers\Reverb\Http\Connection;
use Laravel\Reverb\Servers\Reverb\Http\Response;
use Psr\Http\Message\RequestInterface;
use React\Promise\PromiseInterface;
class ConnectionsController extends Controller
{
/**
* Handle the request.
*/
public function __invoke(RequestInterface $request, Connection $connection, string $appId): PromiseInterface
{
$this->verify($request, $connection, $appId);
return app(MetricsHandler::class)->gather($this->application, 'connections')
->then(fn ($connections) => new Response(['connections' => count($connections)]));
}
}
@@ -0,0 +1,128 @@
<?php
namespace Laravel\Reverb\Protocols\Pusher\Http\Controllers;
use Illuminate\Support\Arr;
use Laravel\Reverb\Application;
use Laravel\Reverb\Contracts\ApplicationProvider;
use Laravel\Reverb\Exceptions\InvalidApplication;
use Laravel\Reverb\Protocols\Pusher\Contracts\ChannelManager;
use Laravel\Reverb\Servers\Reverb\Concerns\ClosesConnections;
use Laravel\Reverb\Servers\Reverb\Http\Connection;
use Psr\Http\Message\RequestInterface;
use Symfony\Component\HttpKernel\Exception\HttpException;
abstract class Controller
{
use ClosesConnections;
/**
* Current application instance.
*/
protected ?Application $application = null;
/**
* Active channels for the application.
*/
protected ?ChannelManager $channels = null;
/**
* The incoming request's body.
*/
protected ?string $body;
/**
* The incoming request's query parameters.
*/
protected array $query = [];
/**
* Verify that the incoming request is valid.
*/
public function verify(RequestInterface $request, Connection $connection, $appId): void
{
parse_str($request->getUri()->getQuery(), $query);
$this->body = $request->getBody()->getContents();
$this->query = $query;
$this->setApplication($appId);
$this->setChannels();
$this->verifySignature($request);
}
/**
* Set the Reverb application instance for the incoming request's application ID.
*
* @throws \Symfony\Component\HttpKernel\Exception\HttpException
*/
protected function setApplication(?string $appId): Application
{
if (! $appId) {
throw new HttpException(400, 'Application ID not provided.');
}
try {
return $this->application = app(ApplicationProvider::class)->findById($appId);
} catch (InvalidApplication $e) {
throw new HttpException(404, 'No matching application for ID ['.$appId.'].');
}
}
/**
* Set the Reverb channel manager instance for the application.
*/
protected function setChannels(): void
{
$this->channels = app(ChannelManager::class)->for($this->application);
}
/**
* Verify the Pusher authentication signature.
*
* @throws \Symfony\Component\HttpKernel\Exception\HttpException
*/
protected function verifySignature(RequestInterface $request): void
{
$params = Arr::except($this->query, [
'auth_signature', 'body_md5', 'appId', 'appKey', 'channelName',
]);
if ($this->body !== '') {
$params['body_md5'] = md5($this->body);
}
ksort($params);
$signature = implode("\n", [
$request->getMethod(),
$request->getUri()->getPath(),
$this->formatQueryParametersForVerification($params),
]);
$signature = hash_hmac('sha256', $signature, $this->application->secret());
$authSignature = $this->query['auth_signature'] ?? '';
if ($signature !== $authSignature) {
throw new HttpException(401, 'Authentication signature invalid.');
}
}
/**
* Format the given parameters into the correct format for signature verification.
*/
protected static function formatQueryParametersForVerification(array $params): string
{
if (! is_array($params)) {
return $params;
}
return collect($params)->map(function ($value, $key) {
if (is_array($value)) {
$value = implode(',', $value);
}
return "{$key}={$value}";
})->implode('&');
}
}
@@ -0,0 +1,96 @@
<?php
namespace Laravel\Reverb\Protocols\Pusher\Http\Controllers;
use Illuminate\Contracts\Validation\Validator;
use Illuminate\Support\Facades\Validator as ValidatorFacade;
use Laravel\Reverb\Protocols\Pusher\Concerns\InteractsWithChannelInformation;
use Laravel\Reverb\Protocols\Pusher\EventDispatcher;
use Laravel\Reverb\Protocols\Pusher\MetricsHandler;
use Laravel\Reverb\Servers\Reverb\Http\Connection;
use Laravel\Reverb\Servers\Reverb\Http\Response;
use Psr\Http\Message\RequestInterface;
use React\Promise\PromiseInterface;
use function React\Promise\all;
class EventsBatchController extends Controller
{
use InteractsWithChannelInformation;
/**
* Handle the request.
*/
public function __invoke(RequestInterface $request, Connection $connection, string $appId): Response|PromiseInterface
{
$this->verify($request, $connection, $appId);
$payload = json_decode($this->body, associative: true, flags: JSON_THROW_ON_ERROR);
$validator = $this->validator($payload);
if ($validator->fails()) {
return new Response($validator->errors(), 422);
}
$items = collect($payload['batch']);
$items = $items->map(function ($item) {
EventDispatcher::dispatch(
$this->application,
[
'event' => $item['name'],
'channel' => $item['channel'],
'data' => $item['data'],
],
isset($item['socket_id']) ? ($this->channels->connections()[$item['socket_id']] ?? null) : null
);
return isset($item['info']) ? app(MetricsHandler::class)->gather(
$this->application,
'channel',
['channel' => $item['channel'], 'info' => $item['info']]
) : [];
});
if ($items->contains(fn ($item) => ! empty($item))) {
return all($items)->then(function ($items) {
return new Response(['batch' => array_map(fn ($item) => (object) $item, $items)]);
});
}
return new Response(['batch' => (object) []]);
}
/**
* Get the info for the given channels.
*
* @return array<string, array<string, int>>
*/
protected function getInfo(string $channel, string $info): array
{
$info = explode(',', $info);
$count = count($this->channels->find($channel)->connections());
$info = [
'user_count' => in_array('user_count', $info) ? $count : null,
'subscription_count' => in_array('subscription_count', $info) ? $count : null,
];
return array_filter($info, fn ($item) => $item !== null);
}
/**
* Create a validator for the incoming request payload.
*/
protected function validator(array $payload): Validator
{
return ValidatorFacade::make($payload, [
'batch' => ['required', 'array'],
'batch.*.name' => ['required', 'string'],
'batch.*.data' => ['required', 'string'],
'batch.*.channel' => ['required_without:channels', 'string'],
'batch.*.socket_id' => ['string'],
'batch.*.info' => ['string'],
]);
}
}
@@ -0,0 +1,73 @@
<?php
namespace Laravel\Reverb\Protocols\Pusher\Http\Controllers;
use Illuminate\Contracts\Validation\Validator;
use Illuminate\Support\Arr;
use Illuminate\Support\Facades\Validator as ValidatorFacade;
use Laravel\Reverb\Protocols\Pusher\Concerns\InteractsWithChannelInformation;
use Laravel\Reverb\Protocols\Pusher\EventDispatcher;
use Laravel\Reverb\Protocols\Pusher\MetricsHandler;
use Laravel\Reverb\Servers\Reverb\Http\Connection;
use Laravel\Reverb\Servers\Reverb\Http\Response;
use Psr\Http\Message\RequestInterface;
use React\Promise\PromiseInterface;
class EventsController extends Controller
{
use InteractsWithChannelInformation;
/**
* Handle the request.
*/
public function __invoke(RequestInterface $request, Connection $connection, string $appId): Response|PromiseInterface
{
$this->verify($request, $connection, $appId);
$payload = json_decode($this->body, associative: true, flags: JSON_THROW_ON_ERROR);
$validator = $this->validator($payload);
if ($validator->fails()) {
return new Response($validator->errors(), 422);
}
$channels = Arr::wrap($payload['channels'] ?? $payload['channel'] ?? []);
if ($except = $payload['socket_id'] ?? null) {
$except = $this->channels->connections()[$except] ?? null;
}
EventDispatcher::dispatch(
$this->application,
[
'event' => $payload['name'],
'channels' => $channels,
'data' => $payload['data'],
],
$except ? $except->connection() : null
);
if (isset($payload['info'])) {
return app(MetricsHandler::class)
->gather($this->application, 'channels', ['info' => $payload['info'], 'channels' => $channels])
->then(fn ($channels) => new Response(['channels' => array_map(fn ($channel) => (object) $channel, $channels)]));
}
return new Response((object) []);
}
/**
* Create a validator for the incoming request payload.
*/
protected function validator(array $payload): Validator
{
return ValidatorFacade::make($payload, [
'name' => ['required', 'string'],
'data' => ['required', 'string'],
'channels' => ['required_without:channel', 'array'],
'channel' => ['required_without:channels', 'string'],
'socket_id' => ['string'],
'info' => ['string'],
]);
}
}
@@ -0,0 +1,18 @@
<?php
namespace Laravel\Reverb\Protocols\Pusher\Http\Controllers;
use Laravel\Reverb\Servers\Reverb\Http\Connection;
use Laravel\Reverb\Servers\Reverb\Http\Response;
use Psr\Http\Message\RequestInterface;
class HealthCheckController extends Controller
{
/**
* Handle the request.
*/
public function __invoke(RequestInterface $request, Connection $connection): Response
{
return new Response((object) ['health' => 'OK']);
}
}
@@ -0,0 +1,70 @@
<?php
namespace Laravel\Reverb\Protocols\Pusher\Http\Controllers;
use Laravel\Reverb\Connection as ReverbConnection;
use Laravel\Reverb\Contracts\ApplicationProvider;
use Laravel\Reverb\Exceptions\InvalidApplication;
use Laravel\Reverb\Protocols\Pusher\Server as PusherServer;
use Laravel\Reverb\Servers\Reverb\Connection;
use Psr\Http\Message\RequestInterface;
use Ratchet\RFC6455\Messaging\FrameInterface;
class PusherController
{
/**
* Create a new controller instance.
*/
public function __construct(protected PusherServer $server, protected ApplicationProvider $applications)
{
//
}
/**
* Invoke the Reverb WebSocket server.
*/
public function __invoke(RequestInterface $request, Connection $connection, string $appKey): void
{
if (! $reverbConnection = $this->connection($request, $connection, $appKey)) {
return;
}
$connection->withMaxMessageSize($reverbConnection->app()->maxMessageSize());
$connection->onMessage(
fn ($message) => $this->server->message($reverbConnection, (string) $message)
);
$connection->onControl(
fn (FrameInterface $message) => $this->server->control($reverbConnection, $message)
);
$connection->onClose(
fn () => $this->server->close($reverbConnection)
);
$connection->openBuffer();
$this->server->open($reverbConnection);
}
/**
* Get the Reverb connection instance for the request.
*/
protected function connection(RequestInterface $request, Connection $connection, string $key): ?ReverbConnection
{
try {
$application = $this->applications->findByKey($key);
} catch (InvalidApplication $e) {
$connection->send('{"event":"pusher:error","data":"{\"code\":4001,\"message\":\"Application does not exist\"}"}');
return $connection->close();
}
return new ReverbConnection(
$connection,
$application,
$request->getHeader('Origin')[0] ?? null
);
}
}
@@ -0,0 +1,39 @@
<?php
namespace Laravel\Reverb\Protocols\Pusher\Http\Controllers;
use Laravel\Reverb\ServerProviderManager;
use Laravel\Reverb\Servers\Reverb\Contracts\PubSubProvider;
use Laravel\Reverb\Servers\Reverb\Http\Connection;
use Laravel\Reverb\Servers\Reverb\Http\Response;
use Psr\Http\Message\RequestInterface;
use React\Promise\PromiseInterface;
class UsersTerminateController extends Controller
{
/**
* Handle the request.
*/
public function __invoke(RequestInterface $request, Connection $connection, string $appId, string $userId): Response|PromiseInterface
{
$this->verify($request, $connection, $appId);
if (app(ServerProviderManager::class)->subscribesToEvents()) {
return app(PubSubProvider::class)->publish([
'type' => 'terminate',
'application' => serialize($this->application),
'payload' => ['user_id' => $userId],
])->then(fn () => new Response((object) []));
}
$connections = collect($this->channels->connections());
$connections->each(function ($connection) use ($userId) {
if ((string) $connection->data()['user_id'] === $userId) {
$connection->disconnect();
}
});
return new Response((object) []);
}
}
@@ -0,0 +1,90 @@
<?php
namespace Laravel\Reverb\Protocols\Pusher\Managers;
use Laravel\Reverb\Contracts\Connection;
use Laravel\Reverb\Protocols\Pusher\Channels\ChannelConnection;
use Laravel\Reverb\Protocols\Pusher\Contracts\ChannelConnectionManager;
class ArrayChannelConnectionManager implements ChannelConnectionManager
{
/**
* The channel name.
*/
protected string $name;
/**
* The underlying connections.
*
* @var array<string, \Laravel\Reverb\Protocols\Pusher\Channels\ChannelConnection>
*/
protected $connections = [];
/**
* Get a channel connection manager for the given channel name.
*/
public function for(string $name): ChannelConnectionManager
{
$this->name = $name;
return $this;
}
/**
* Add a connection.
*/
public function add(Connection $connection, array $data): void
{
$this->connections[$connection->id()] = new ChannelConnection($connection, $data);
}
/**
* Remove a connection.
*/
public function remove(Connection $connection): void
{
unset($this->connections[$connection->id()]);
}
/**
* Find a connection.
*/
public function find(Connection $connection): ?ChannelConnection
{
return $this->findById($connection->id());
}
/**
* Find a connection by its ID.
*/
public function findById(string $id): ?ChannelConnection
{
return $this->connections[$id] ?? null;
}
/**
* Get all of the connections.
*
* @return array<string, \Laravel\Reverb\Protocols\Pusher\Channels\ChannelConnection>
*/
public function all(): array
{
return $this->connections;
}
/**
* Determine whether any connections remain on the channel.
*/
public function isEmpty(): bool
{
return empty($this->connections);
}
/**
* Flush the channel connection manager.
*/
public function flush(): void
{
$this->connections = [];
}
}
@@ -0,0 +1,155 @@
<?php
namespace Laravel\Reverb\Protocols\Pusher\Managers;
use Illuminate\Support\Arr;
use Laravel\Reverb\Application;
use Laravel\Reverb\Concerns\InteractsWithApplications;
use Laravel\Reverb\Contracts\ApplicationProvider;
use Laravel\Reverb\Contracts\Connection;
use Laravel\Reverb\Events\ChannelCreated;
use Laravel\Reverb\Events\ChannelRemoved;
use Laravel\Reverb\Protocols\Pusher\Channels\Channel;
use Laravel\Reverb\Protocols\Pusher\Channels\ChannelBroker;
use Laravel\Reverb\Protocols\Pusher\Contracts\ChannelManager as ChannelManagerInterface;
class ArrayChannelManager implements ChannelManagerInterface
{
use InteractsWithApplications;
/**
* The underlying array of applications and their channels.
*
* @var array<string, array<string, array<string, \Laravel\Reverb\Protocols\Pusher\Channels\Channel>>>
*/
protected $applications = [];
/**
* The application instance.
*
* @var \Laravel\Reverb\Application
*/
protected $application;
/**
* Get the application instance.
*/
public function app(): ?Application
{
return $this->application;
}
/**
* Get all the channels.
*
* @return array<string, \Laravel\Reverb\Protocols\Pusher\Channels\Channel>
*/
public function all(): array
{
return $this->channels();
}
/**
* Determine whether the given channel exists.
*/
public function exists(string $channel): bool
{
return isset($this->applications[$this->application->id()][$channel]);
}
/**
* Find the given channel
*/
public function find(string $channel): ?Channel
{
return $this->channels($channel);
}
/**
* Find the given channel or create it if it doesn't exist.
*/
public function findOrCreate(string $channelName): Channel
{
if ($channel = $this->find($channelName)) {
return $channel;
}
$channel = ChannelBroker::create($channelName);
$this->applications[$this->application->id()][$channel->name()] = $channel;
ChannelCreated::dispatch($channel);
return $channel;
}
/**
* Get all of the connections for the given channels.
*
* @return array<string, \Laravel\Reverb\Protocols\Pusher\Channels\ChannelConnection>
*/
public function connections(?string $channel = null): array
{
$channels = Arr::wrap($this->channels($channel));
return array_reduce($channels, function ($carry, $channel) {
return $carry + $channel->connections();
}, []);
}
/**
* Unsubscribe from all channels.
*/
public function unsubscribeFromAll(Connection $connection): void
{
foreach ($this->channels() as $channel) {
$channel->unsubscribe($connection);
}
}
/**
* Remove the given channel.
*/
public function remove(Channel $channel): void
{
unset($this->applications[$this->application->id()][$channel->name()]);
ChannelRemoved::dispatch($channel);
}
/**
* Get the given channel.
*/
public function channel(string $channel): ?Channel
{
return $this->channels($channel);
}
/**
* Get the channels for the application.
*
* @return \Laravel\Reverb\Protocols\Pusher\Channels\Channel|array<string, \Laravel\Reverb\Protocols\Pusher\Channels\Channel>|null
*/
public function channels(?string $channel = null): Channel|array|null
{
$channels = $this->applications[$this->application->id()] ?? [];
if (isset($channel)) {
return $channels[$channel] ?? null;
}
return $channels;
}
/**
* Flush the channel manager repository.
*/
public function flush(): void
{
app(ApplicationProvider::class)
->all()
->each(function (Application $application) {
$this->applications[$application->id()] = [];
});
}
}
@@ -0,0 +1,11 @@
<?php
namespace Laravel\Reverb\Protocols\Pusher;
enum MetricType: string
{
case CONNECTIONS = 'connections';
case CHANNEL = 'channel';
case CHANNELS = 'channels';
case CHANNEL_USERS = 'channel_users';
}
@@ -0,0 +1,266 @@
<?php
namespace Laravel\Reverb\Protocols\Pusher;
use Illuminate\Support\Str;
use Laravel\Reverb\Application;
use Laravel\Reverb\Protocols\Pusher\Concerns\InteractsWithChannelInformation;
use Laravel\Reverb\Protocols\Pusher\Contracts\ChannelManager;
use Laravel\Reverb\ServerProviderManager;
use Laravel\Reverb\Servers\Reverb\Contracts\PubSubProvider;
use React\Promise\Deferred;
use React\Promise\PromiseInterface;
use function React\Promise\Timer\timeout;
class MetricsHandler
{
use InteractsWithChannelInformation;
/**
* The metrics being gathered.
*
* @var array<string, PendingMetric>
*/
protected array $metrics = [];
/**
* Create an instance of the metrics handler.
*/
public function __construct(
protected ServerProviderManager $serverProviderManager,
protected ChannelManager $channels,
protected PubSubProvider $pubSubProvider
) {
//
}
/**
* Gather the metrics for the given type.
*/
public function gather(Application $application, string $type, array $options = []): PromiseInterface
{
$metric = new PendingMetric(
Str::random(10),
$application,
MetricType::from($type),
$options
);
return $this->serverProviderManager->subscribesToEvents()
? $this->gatherMetricsFromSubscribers($metric)
: $this->promise($this->get($metric));
}
/**
* Get the metrics for the given type.
*/
public function get(PendingMetric $metric): array
{
return match ($metric->type()) {
MetricType::CHANNEL => $this->channel($metric),
MetricType::CHANNELS => $this->channels($metric),
MetricType::CHANNEL_USERS => $this->channelUsers($metric),
MetricType::CONNECTIONS => $this->connections($metric),
default => [],
};
}
/**
* Get the channel for the given application.
*/
protected function channel(PendingMetric $metric): array
{
return $this->info($metric->application(), $metric->option('channel'), $metric->option('info', ''));
}
/**
* Get the channels for the given application.
*/
protected function channels(PendingMetric $metric): array
{
if ($metric->option('channels')) {
return $this->infoForChannels($metric->application(), $metric->option('channels'), $metric->option('info', ''));
}
$channels = collect($this->channels->for($metric->application())->all());
if ($filter = ($metric->option('filter', false))) {
$channels = $channels->filter(fn ($channel) => Str::startsWith($channel->name(), $filter));
}
$channels = $channels->filter(fn ($channel) => count($channel->connections()) > 0);
return $this->infoForChannels(
$metric->application(),
$channels->all(),
$metric->option('info', '')
);
}
/**
* Get the channel users for the given application.
*/
protected function channelUsers(PendingMetric $metric): array
{
$channel = $this->channels->for($metric->application())->find($metric->option('channel'));
if (! $channel) {
return [];
}
return collect($channel->connections())
->map(fn ($connection) => $connection->data())
->unique('user_id')
->map(fn ($data) => ['id' => $data['user_id']])
->values()
->all();
}
/**
* Get the connections for the given application.
*/
protected function connections(PendingMetric $metric): array
{
return $this->channels->for($metric->application())->connections();
}
/**
* Gather metrics from all subscribers for the given type.
*/
protected function gatherMetricsFromSubscribers(PendingMetric $metric): PromiseInterface
{
$this->metrics[$metric->key()] = $metric;
$deferred = $this->listenForMetrics($metric);
$this->requestMetricsFromSubscribers($metric);
return timeout($deferred->promise(), 10)->then(
fn ($metrics) => $metrics,
fn () => $this->metrics[$metric->key()]?->resolve() ?? [],
)->then(
fn ($metrics) => $this->mergeSubscriberMetrics($metrics, $metric->type())
)->finally(
fn () => $this->stopListening($metric)
);
}
/**
* Request metrics from all subscribers.
*/
protected function requestMetricsFromSubscribers(PendingMetric $metric): void
{
$this->pubSubProvider->publish([
'type' => 'metrics',
'payload' => serialize($metric),
])->then(function ($total) use ($metric) {
$metric->setSubscriberCount($total);
});
}
/**
* Merge the given metrics into a single result set.
*/
protected function mergeSubscriberMetrics(array $metrics, MetricType $type): array
{
return match ($type) {
MetricType::CONNECTIONS => array_reduce($metrics, fn ($carry, $item) => array_merge($carry, $item), []),
MetricType::CHANNELS => $this->mergeChannels($metrics),
MetricType::CHANNEL => $this->mergeChannel($metrics),
MetricType::CHANNEL_USERS => collect($metrics)->flatten(1)->unique()->all(),
default => [],
};
}
/**
* Merge multiple channel instances into a single set.
*/
protected function mergeChannel(array $metrics): array
{
return collect($metrics)
->reduce(function ($carry, $item) {
collect($item)->each(fn ($value, $key) => $carry->put($key, match ($key) {
'occupied' => $carry->get($key, false) || $value,
'user_count' => $carry->get($key, 0) + $value,
'subscription_count' => $carry->get($key, 0) + $value,
default => $value,
}));
return $carry;
}, collect())
->all();
}
/**
* Merge multiple sets of channel instances into a single result set.
*/
protected function mergeChannels(array $metrics): array
{
return collect($metrics)
->reduce(function ($carry, $item) {
collect($item)->each(function ($data, $channel) use ($carry) {
$metrics = $carry->get($channel, []);
$metrics[] = $data;
$carry->put($channel, $metrics);
});
return $carry;
}, collect())
->map(fn ($metrics) => $this->mergeChannel($metrics))
->all();
}
/**
* Listen for metrics from subscribers.
*/
protected function listenForMetrics(PendingMetric $metric): Deferred
{
$deferred = new Deferred;
$this->pubSubProvider->on($metric->key(), function ($payload) use ($metric, $deferred) {
$pending = $this->metrics[$metric->key()];
$pending->append($payload['payload']);
if ($pending->resolvable()) {
$deferred->resolve($pending->resolve());
}
});
return $deferred;
}
/**
* Publish the metrics for the given type.
*/
public function publish(PendingMetric $metric): void
{
$this->pubSubProvider->publish([
'type' => $metric->key(),
'payload' => $this->get($metric),
]);
}
/**
* Stop listening for the given metric.
*/
protected function stopListening(PendingMetric $metric): void
{
unset($this->metrics[$metric->key()]);
$this->pubSubProvider->stopListening($metric->key());
}
/**
* Create a promise to resolve the given value.
*/
protected function promise(mixed $value): PromiseInterface
{
$deferred = new Deferred;
$promise = $deferred->promise();
$deferred->resolve($value);
return $promise;
}
}
@@ -0,0 +1,102 @@
<?php
namespace Laravel\Reverb\Protocols\Pusher;
use Laravel\Reverb\Application;
class PendingMetric
{
/**
* The number of subscribers for the metric.
*/
protected ?int $subscribers = null;
/**
* The data for the metric.
*/
protected array $data = [];
/**
* Instantiate a new pending metric.
*/
public function __construct(
protected string $key,
protected Application $application,
protected MetricType $type,
protected array $options = [],
) {
//
}
/**
* Get the metric key.
*/
public function key(): string
{
return $this->key;
}
/**
* Get the metric type.
*/
public function type(): MetricType
{
return $this->type;
}
/**
* Get the application for the metric.
*/
public function application(): Application
{
return $this->application;
}
/**
* Get an option for the metric.
*/
public function option(string $key, mixed $default = null): mixed
{
return $this->options[$key] ?? $default;
}
/**
* Get the options for the metric.
*/
public function options(): array
{
return $this->options;
}
/**
* Set the subscriber count for the metric.
*/
public function setSubscriberCount(int $count): void
{
$this->subscribers = $count;
}
/**
* Append data to the metric.
*/
public function append(array $data): void
{
$this->data[] = $data;
}
/**
* Check if the metric is resolvable.
*/
public function resolvable(): bool
{
return $this->subscribers !== null && count($this->data) === $this->subscribers;
}
/**
* Resolve the data for the metric.
*/
public function resolve(): array
{
return $this->data;
}
}
@@ -0,0 +1,78 @@
<?php
namespace Laravel\Reverb\Protocols\Pusher;
use Laravel\Reverb\Application;
use Laravel\Reverb\Protocols\Pusher\Contracts\ChannelManager;
use Laravel\Reverb\Servers\Reverb\Contracts\PubSubIncomingMessageHandler;
class PusherPubSubIncomingMessageHandler implements PubSubIncomingMessageHandler
{
protected array $events = [];
/**
* Handle an incoming message from the PubSub provider.
*/
public function handle(string $payload): void
{
$event = json_decode($payload, associative: true, flags: JSON_THROW_ON_ERROR);
$this->processEventListeners($event);
$application = unserialize($event['application'] ?? null, ['allowed_classes' => [Application::class]]);
$except = isset($event['socket_id']) ?
app(ChannelManager::class)->for($application)->connections()[$event['socket_id']] ?? null
: null;
match ($event['type'] ?? null) {
'message' => EventDispatcher::dispatchSynchronously(
$application,
$event['payload'],
$except?->connection()
),
'metrics' => app(MetricsHandler::class)->publish(
unserialize($event['payload'], ['allowed_classes' => [
Application::class, PendingMetric::class, MetricType::class,
]])
),
'terminate' => collect(app(ChannelManager::class)->for($application)->connections())
->each(function ($connection) use ($event) {
if ((string) $connection->data()['user_id'] === $event['payload']['user_id']) {
$connection->disconnect();
}
}),
default => null,
};
}
/**
* Process the given event.
*/
protected function processEventListeners(array $event): void
{
foreach ($this->events as $eventName => $listeners) {
if (($event['type'] ?? null) === $eventName) {
foreach ($listeners as $listener) {
$listener($event);
}
}
}
}
/**
* Listen for the given event.
*/
public function listen(string $event, callable $callback): void
{
$this->events[$event][] = $callback;
}
/**
* Stop listening for the given event.
*/
public function stopListening(string $event): void
{
unset($this->events[$event]);
}
}
+178
View File
@@ -0,0 +1,178 @@
<?php
namespace Laravel\Reverb\Protocols\Pusher;
use Exception;
use Illuminate\Support\Facades\Validator;
use Illuminate\Support\Str;
use Laravel\Reverb\Contracts\Connection;
use Laravel\Reverb\Events\MessageReceived;
use Laravel\Reverb\Loggers\Log;
use Laravel\Reverb\Protocols\Pusher\Contracts\ChannelManager;
use Laravel\Reverb\Protocols\Pusher\Exceptions\ConnectionLimitExceeded;
use Laravel\Reverb\Protocols\Pusher\Exceptions\InvalidOrigin;
use Laravel\Reverb\Protocols\Pusher\Exceptions\PusherException;
use Ratchet\RFC6455\Messaging\Frame;
use Ratchet\RFC6455\Messaging\FrameInterface;
use Throwable;
class Server
{
/**
* Create a new server instance.
*/
public function __construct(protected ChannelManager $channels, protected EventHandler $handler)
{
//
}
/**
* Handle the a client connection.
*/
public function open(Connection $connection): void
{
try {
$this->ensureWithinConnectionLimit($connection);
$this->verifyOrigin($connection);
$connection->touch();
$this->handler->handle($connection, 'pusher:connection_established');
Log::info('Connection Established', $connection->id());
} catch (Exception $e) {
$this->error($connection, $e);
}
}
/**
* Handle a new message received by the connected client.
*/
public function message(Connection $from, string $message): void
{
Log::info('Message Received', $from->id());
Log::message($message);
$from->touch();
try {
$event = json_decode($message, associative: true, flags: JSON_THROW_ON_ERROR);
if (Str::isJson($event['data'] ?? null)) {
$event['data'] = json_decode($event['data'], associative: true, flags: JSON_THROW_ON_ERROR);
}
Validator::make($event, ['event' => ['required', 'string']])->validate();
match (Str::startsWith($event['event'], 'pusher:')) {
true => $this->handler->handle(
$from,
$event['event'],
empty($event['data']) ? [] : $event['data'],
),
default => ClientEvent::handle($from, $event)
};
Log::info('Message Handled', $from->id());
MessageReceived::dispatch($from, $message);
} catch (Throwable $e) {
$this->error($from, $e);
}
}
/**
* Handle a low-level WebSocket control frame.
*/
public function control(Connection $from, FrameInterface $message): void
{
Log::info('Control Frame Received', $from->id());
Log::message($message);
$from->setUsesControlFrames();
if (in_array($message->getOpcode(), [Frame::OP_PING, Frame::OP_PONG], strict: true)) {
$from->touch();
}
}
/**
* Handle a client disconnection.
*/
public function close(Connection $connection): void
{
$this->channels
->for($connection->app())
->unsubscribeFromAll($connection);
$connection->disconnect();
Log::info('Connection Closed', $connection->id());
}
/**
* Handle an error.
*/
public function error(Connection $connection, Throwable $exception): void
{
if ($exception instanceof PusherException) {
$connection->send(json_encode($exception->payload()));
Log::error('Message from '.$connection->id().' resulted in a pusher error');
Log::info($exception->getMessage());
return;
}
$connection->send(json_encode([
'event' => 'pusher:error',
'data' => json_encode([
'code' => 4200,
'message' => 'Invalid message format',
]),
]));
Log::error('Message from '.$connection->id().' resulted in an unknown error');
Log::info($exception->getMessage());
}
/**
* Ensure the server is within the connection limit.
*/
protected function ensureWithinConnectionLimit(Connection $connection): void
{
if (! $connection->app()->hasMaxConnectionLimit()) {
return;
}
$connections = $this->channels->for($connection->app())->connections();
if (count($connections) >= $connection->app()->maxConnections()) {
throw new ConnectionLimitExceeded;
}
}
/**
* Verify the origin of the connection.
*
* @throws \Laravel\Reverb\Exceptions\InvalidOrigin
*/
protected function verifyOrigin(Connection $connection): void
{
$allowedOrigins = $connection->app()->allowedOrigins();
if (in_array('*', $allowedOrigins)) {
return;
}
$origin = parse_url($connection->origin(), PHP_URL_HOST);
foreach ($allowedOrigins as $allowedOrigin) {
if (Str::is($allowedOrigin, $origin)) {
return;
}
}
throw new InvalidOrigin;
}
}
@@ -0,0 +1,52 @@
<?php
namespace Laravel\Reverb\Pulse\Livewire\Concerns;
use Carbon\CarbonInterval;
trait HasRate
{
/**
* The message send rate.
*/
protected function rate(?float $count): ?float
{
return $count ? round($count * $this->rateMultiplier(), 2) : null;
}
/**
* The rate multiplier.
*/
protected function rateMultiplier(): float
{
return with(match ($this->period) {
'6_hours' => CarbonInterval::minute(),
'24_hours' => CarbonInterval::hour(),
'7_days' => CarbonInterval::day(),
default => CarbonInterval::second(),
}, fn ($period) => $period->totalSeconds > $this->secondsPerBucket()
? $this->secondsPerBucket() / $period->totalSeconds
: $period->totalSeconds / $this->secondsPerBucket());
}
/**
* The seconds per bucket.
*/
protected function secondsPerBucket(): float
{
return $this->periodAsInterval()->totalSeconds / $maxDataPoints = 60;
}
/**
* The rate unit.
*/
protected function rateUnit(): string
{
return match ($this->period) {
'6_hours' => 'minute',
'24_hours' => 'hour',
'7_days' => 'day',
default => 'second',
};
}
}
@@ -0,0 +1,67 @@
<?php
namespace Laravel\Reverb\Pulse\Livewire;
use Illuminate\Support\Facades\Config;
use Illuminate\Support\Facades\Request;
use Illuminate\Support\Facades\View;
use Illuminate\Support\HtmlString;
use Laravel\Pulse\Livewire\Card;
use Laravel\Pulse\Livewire\Concerns\HasPeriod;
use Laravel\Pulse\Livewire\Concerns\RemembersQueries;
use Laravel\Reverb\Pulse\Recorders\ReverbConnections;
use Livewire\Attributes\Lazy;
class Connections extends Card
{
use HasPeriod, RemembersQueries;
/**
* The graph colors.
*/
public array $colors = [
'avg' => '#10b981',
'max' => '#9333ea',
];
/**
* Render the component.
*/
#[Lazy]
public function render()
{
[$connections, $time, $runAt] = $this->remember(function () {
return with($this->graph(['reverb_connections'], 'max'), function ($max) {
return $this->graph(['reverb_connections'], 'avg')->map(fn ($readings, $app) => collect([
'reverb_connections:avg' => $readings['reverb_connections'],
'reverb_connections:max' => $max[$app]['reverb_connections'],
]));
});
});
if (Request::hasHeader('X-Livewire')) {
$this->dispatch('reverb-connections-chart-update', connections: $connections);
}
return View::make('reverb::livewire.connections', [
'connections' => $connections,
'time' => $time,
'runAt' => $runAt,
'config' => Config::get('pulse.recorders.'.ReverbConnections::class),
]);
}
/**
* Define any CSS that should be loaded for the component.
*
* @return string|\Illuminate\Contracts\Support\Htmlable|array<int, string|\Illuminate\Contracts\Support\Htmlable>|null
*/
protected function css(): HtmlString
{
return new HtmlString(
'<style>'.
collect($this->colors)->map(fn ($color) => '.bg-\\[\\'.$color.'\\]{background-color:'.$color.'}')->join('').
'</style>'
);
}
}
+70
View File
@@ -0,0 +1,70 @@
<?php
namespace Laravel\Reverb\Pulse\Livewire;
use Illuminate\Support\Facades\Config;
use Illuminate\Support\Facades\Request;
use Illuminate\Support\Facades\View;
use Illuminate\Support\HtmlString;
use Laravel\Pulse\Livewire\Card;
use Laravel\Pulse\Livewire\Concerns\HasPeriod;
use Laravel\Pulse\Livewire\Concerns\RemembersQueries;
use Laravel\Reverb\Pulse\Recorders\ReverbMessages;
use Livewire\Attributes\Lazy;
class Messages extends Card
{
use Concerns\HasRate,
HasPeriod,
RemembersQueries;
/**
* The graph colors.
*/
public array $colors = [
'received' => '#10b981',
'received:per_rate' => '#78d7b3',
'sent' => '#9333ea',
'sent:per_rate' => '#bc81f1',
];
/**
* Render the component.
*/
#[Lazy]
public function render()
{
[$all, $time, $runAt] = $this->remember(fn () => [
$readings = $this->graph(['reverb_message:sent', 'reverb_message:received'], 'count'),
$readings->map->map(fn ($values) => $values->map($this->rate(...))),
]);
[$messages, $messagesRate] = $all;
if (Request::hasHeader('X-Livewire')) {
$this->dispatch('reverb-messages-chart-update', messages: $messages, messagesRate: $messagesRate);
}
return View::make('reverb::livewire.messages', [
'messages' => $messages,
'messagesRate' => $messagesRate,
'time' => $time,
'runAt' => $runAt,
'config' => Config::get('pulse.recorders.'.ReverbMessages::class),
]);
}
/**
* Define any CSS that should be loaded for the component.
*
* @return string|\Illuminate\Contracts\Support\Htmlable|array<int, string|\Illuminate\Contracts\Support\Htmlable>|null
*/
protected function css(): HtmlString
{
return new HtmlString(
'<style>'.
collect($this->colors)->map(fn ($color) => '.bg-\\[\\'.$color.'\\]{background-color:'.$color.'}')->join('').
'</style>'
);
}
}
@@ -0,0 +1,58 @@
<?php
namespace Laravel\Reverb\Pulse\Recorders;
use Illuminate\Broadcasting\BroadcastManager;
use Illuminate\Contracts\Foundation\Application as Container;
use Laravel\Pulse\Events\IsolatedBeat;
use Laravel\Pulse\Pulse;
use Laravel\Pulse\Recorders\Concerns\Sampling;
use Laravel\Reverb\Application;
use Laravel\Reverb\Contracts\ApplicationProvider;
class ReverbConnections
{
use Sampling;
/**
* The events to listen for.
*
* @var class-string
*/
public string $listen = IsolatedBeat::class;
/**
* Create a new recorder instance.
*/
public function __construct(
protected Pulse $pulse,
protected BroadcastManager $broadcast,
protected Container $app,
) {
//
}
/**
* Record the connection count.
*/
public function record(IsolatedBeat $event): void
{
if ($event->time->second % 15 !== 0) {
return;
}
$this->app->make(ApplicationProvider::class)->all()
->each(function (Application $app) use ($event) {
$connections = $this->broadcast->pusher($app->toArray())
->get('/connections')
->connections;
$this->pulse->record(
type: 'reverb_connections',
key: $app->id(),
value: $connections,
timestamp: $event->time->getTimestamp(),
)->avg()->max()->onlyBuckets();
});
}
}
@@ -0,0 +1,54 @@
<?php
namespace Laravel\Reverb\Pulse\Recorders;
use Carbon\CarbonImmutable;
use Illuminate\Config\Repository;
use Laravel\Pulse\Pulse;
use Laravel\Pulse\Recorders\Concerns\Sampling;
use Laravel\Reverb\Events\MessageReceived;
use Laravel\Reverb\Events\MessageSent;
class ReverbMessages
{
use Sampling;
/**
* The events to listen for.
*
* @var list<class-string>
*/
public array $listen = [
MessageSent::class,
MessageReceived::class,
];
/**
* Create a new recorder instance.
*/
public function __construct(
protected Pulse $pulse,
protected Repository $config
) {
//
}
/**
* Record the message.
*/
public function record(MessageSent|MessageReceived $event): void
{
if (! $this->shouldSample()) {
return;
}
$this->pulse->record(
type: 'reverb_message:'.match ($event::class) {
MessageSent::class => 'sent',
MessageReceived::class => 'received',
},
key: $event->connection->app()->id(),
timestamp: CarbonImmutable::now()->getTimestamp(),
)->onlyBuckets()->count();
}
}
+58
View File
@@ -0,0 +1,58 @@
<?php
namespace Laravel\Reverb;
use Illuminate\Support\ServiceProvider;
use Laravel\Reverb\Console\Commands\InstallCommand;
use Laravel\Reverb\Contracts\Logger;
use Laravel\Reverb\Loggers\NullLogger;
use Laravel\Reverb\Pulse\Livewire;
use Livewire\LivewireManager;
class ReverbServiceProvider extends ServiceProvider
{
/**
* Register any application services.
*/
public function register(): void
{
$this->mergeConfigFrom(
__DIR__.'/../config/reverb.php', 'reverb'
);
$this->app->instance(Logger::class, new NullLogger);
$this->app->singleton(ServerProviderManager::class);
$this->app->make(ServerProviderManager::class)->register();
}
/**
* Bootstrap any application services.
*/
public function boot(): void
{
if ($this->app->runningInConsole()) {
$this->commands(InstallCommand::class);
$this->publishes([
__DIR__.'/../config/reverb.php' => config_path('reverb.php'),
], ['reverb', 'reverb-config']);
if (method_exists($this, 'reloads')) {
$this->reloads('reverb:restart', 'reverb');
}
}
if ($this->app->bound(\Laravel\Pulse\Pulse::class)) {
$this->loadViewsFrom(__DIR__.'/../resources/views', 'reverb');
$this->callAfterResolving('livewire', function (LivewireManager $livewire) {
$livewire->component('reverb.messages', Livewire\Messages::class);
$livewire->component('reverb.connections', Livewire\Connections::class);
});
}
$this->app->make(ServerProviderManager::class)->boot();
}
}
+37
View File
@@ -0,0 +1,37 @@
<?php
namespace Laravel\Reverb;
use Illuminate\Contracts\Foundation\Application;
use Illuminate\Support\Manager;
use Laravel\Reverb\Servers\Reverb\ReverbServerProvider;
class ServerProviderManager extends Manager
{
/**
* Create a new server manager instance.
*/
public function __construct(protected Application $app)
{
parent::__construct($app);
}
/**
* Creates the Reverb driver.
*/
public function createReverbDriver(): ReverbServerProvider
{
return new ReverbServerProvider(
$this->app,
$this->config->get('reverb.servers.reverb', [])
);
}
/**
* Get the default driver name.
*/
public function getDefaultDriver(): string
{
return $this->config->get('reverb.default', 'reverb');
}
}
@@ -0,0 +1,22 @@
<?php
namespace Laravel\Reverb\Servers\Reverb\Concerns;
use GuzzleHttp\Psr7\Message;
use GuzzleHttp\Psr7\Response;
use Laravel\Reverb\Servers\Reverb\Http\Connection;
trait ClosesConnections
{
/**
* Close the connection.
*/
protected function close(Connection $connection, int $statusCode = 400, string $message = '', array $headers = []): void
{
$response = new Response($statusCode, $headers, $message);
$connection->send(Message::toString($response));
$connection->close();
}
}
+159
View File
@@ -0,0 +1,159 @@
<?php
namespace Laravel\Reverb\Servers\Reverb;
use Evenement\EventEmitter;
use Laravel\Reverb\Contracts\WebSocketConnection;
use Laravel\Reverb\Servers\Reverb\Http\Connection as HttpConnection;
use Ratchet\RFC6455\Messaging\CloseFrameChecker;
use Ratchet\RFC6455\Messaging\DataInterface;
use Ratchet\RFC6455\Messaging\Frame;
use Ratchet\RFC6455\Messaging\FrameInterface;
use Ratchet\RFC6455\Messaging\MessageBuffer;
class Connection extends EventEmitter implements WebSocketConnection
{
/**
* The message buffer.
*
* @var \Ratchet\RFC6455\Messaging\MessageBuffer
*/
protected $buffer;
/**
* The message handler.
*
* @var ?callable
*/
protected $onMessage;
/**
* The control frame handler.
*
* @var ?callable
*/
protected $onControl;
/**
* The connection close handler.
*
* @var ?callable
*/
protected $onClose;
/**
* The maximum number of allowed bytes for each message.
*
* @var int
*/
protected $maxMessageSize;
/**
* Create a new websocket connection instance.
*/
public function __construct(public HttpConnection $connection)
{
//
}
/**
* Undocumented function
*/
public function openBuffer(): void
{
$this->buffer = new MessageBuffer(
new CloseFrameChecker,
maxMessagePayloadSize: $this->maxMessageSize,
onMessage: $this->onMessage ?: fn () => null,
onControl: fn (FrameInterface $message) => $this->control($message),
sender: [$this->connection, 'send']
);
$this->connection->on('data', [$this->buffer, 'onData']);
$this->connection->on('close', $this->onClose ?: fn () => null);
}
/**
* Send a message to the connection.
*/
public function send(mixed $message): void
{
$this->connection->send(
$message instanceof DataInterface ?
$message->getContents() :
(new Frame($message))->getContents()
);
}
/**
* Handle control frames.
*/
public function control(FrameInterface $message): void
{
if ($this->onControl) {
($this->onControl)($message);
}
match ($message->getOpcode()) {
Frame::OP_PING => $this->send(new Frame($message->getPayload(), opcode: Frame::OP_PONG)),
Frame::OP_PONG => fn () => null,
Frame::OP_CLOSE => $this->close($message),
};
}
/**
* Set the message handler.
*/
public function onMessage(callable $callback): void
{
$this->onMessage = $callback;
}
/**
* Set the control frame handler.
*/
public function onControl(callable $callback): void
{
$this->onControl = $callback;
}
/**
* Set the close handler.
*/
public function onClose(callable $callback): void
{
$this->onClose = $callback;
}
/**
* Set the maximum number of allowed bytes for each message from the client.
*/
public function withMaxMessageSize(int $size): void
{
$this->maxMessageSize = $size;
}
/**
* Close the connection.
*/
public function close(mixed $message = null): void
{
if ($message) {
$frame = $message instanceof FrameInterface ?
$message :
new Frame($message, opcode: Frame::OP_CLOSE);
$this->send($frame);
}
$this->connection->close();
}
/**
* Get the raw socket connection identifier.
*/
public function id(): int
{
return $this->connection->id();
}
}
@@ -0,0 +1,38 @@
<?php
namespace Laravel\Reverb\Servers\Reverb\Console\Commands;
use Illuminate\Console\Command;
use Illuminate\Support\Facades\Cache;
use Illuminate\Support\InteractsWithTime;
use Symfony\Component\Console\Attribute\AsCommand;
#[AsCommand(name: 'reverb:restart')]
class RestartServer extends Command
{
use InteractsWithTime;
/**
* The name and signature of the console command.
*
* @var string
*/
protected $signature = 'reverb:restart';
/**
* The console command description.
*
* @var string
*/
protected $description = 'Restart the Reverb server';
/**
* Execute the console command.
*/
public function handle(): void
{
Cache::forever('laravel:reverb:restart', $this->currentTime());
$this->components->info('Broadcasting Reverb restart signal.');
}
}
@@ -0,0 +1,199 @@
<?php
namespace Laravel\Reverb\Servers\Reverb\Console\Commands;
use Illuminate\Console\Command;
use Illuminate\Support\Facades\Cache;
use Laravel\Reverb\Application;
use Laravel\Reverb\Contracts\ApplicationProvider;
use Laravel\Reverb\Contracts\Logger;
use Laravel\Reverb\Jobs\PingInactiveConnections;
use Laravel\Reverb\Jobs\PruneStaleConnections;
use Laravel\Reverb\Loggers\CliLogger;
use Laravel\Reverb\Protocols\Pusher\Contracts\ChannelManager;
use Laravel\Reverb\ServerProviderManager;
use Laravel\Reverb\Servers\Reverb\Contracts\PubSubProvider;
use Laravel\Reverb\Servers\Reverb\Factory as ServerFactory;
use Laravel\Reverb\Servers\Reverb\Http\Server;
use React\EventLoop\Loop;
use React\EventLoop\LoopInterface;
use Symfony\Component\Console\Attribute\AsCommand;
use Symfony\Component\Console\Command\SignalableCommandInterface;
#[AsCommand(name: 'reverb:start')]
class StartServer extends Command implements SignalableCommandInterface
{
/**
* The name and signature of the console command.
*
* @var string
*/
protected $signature = 'reverb:start
{--host= : The IP address the server should bind to}
{--port= : The port the server should listen on}
{--path= : The path the server should prefix to all routes}
{--hostname= : The hostname the server is accessible from}
{--debug : Indicates whether debug messages should be displayed in the terminal}';
/**
* The console command description.
*
* @var string
*/
protected $description = 'Start the Reverb server';
/**
* Execute the console command.
*/
public function handle(): void
{
if ($this->option('debug')) {
$this->laravel->instance(Logger::class, new CliLogger($this->output));
}
$config = $this->laravel['config']['reverb.servers.reverb'];
$loop = Loop::get();
$server = ServerFactory::make(
$host = $this->option('host') ?: $config['host'],
$port = $this->option('port') ?: $config['port'],
$path = $this->option('path') ?: $config['path'] ?? '',
$hostname = $this->option('hostname') ?: $config['hostname'],
$config['max_request_size'] ?? 10_000,
$config['options'] ?? [],
loop: $loop
);
$this->ensureHorizontalScalability($loop);
$this->ensureStaleConnectionsAreCleaned($loop);
$this->ensureRestartCommandIsRespected($server, $loop, $host, $port);
$this->ensurePulseEventsAreCollected($loop, $config['pulse_ingest_interval']);
$this->ensureTelescopeEntriesAreCollected($loop, $config['telescope_ingest_interval'] ?? 15);
$this->components->info('Starting '.($server->isSecure() ? 'secure ' : '')."server on {$host}:{$port}{$path}".(($hostname && $hostname !== $host) ? " ({$hostname})" : ''));
$server->start();
}
/**
* Ensure that horizontal scalability via broadcasting is enabled if configured.
*/
protected function ensureHorizontalScalability(LoopInterface $loop): void
{
if ($this->laravel->make(ServerProviderManager::class)->driver('reverb')->subscribesToEvents()) {
$this->laravel->make(PubSubProvider::class)->connect($loop);
}
}
/**
* Use the event loop to schedule periodic cleanup of connections.
*/
protected function ensureStaleConnectionsAreCleaned(LoopInterface $loop): void
{
$loop->addPeriodicTimer(60, function () {
PruneStaleConnections::dispatch();
PingInactiveConnections::dispatch();
});
}
/**
* Check to see whether the restart signal has been sent.
*/
protected function ensureRestartCommandIsRespected(Server $server, LoopInterface $loop, string $host, string $port): void
{
$lastRestart = Cache::get('laravel:reverb:restart');
$loop->addPeriodicTimer(5, function () use ($server, $host, $port, $lastRestart) {
if ($lastRestart === Cache::get('laravel:reverb:restart')) {
return;
}
$this->gracefullyDisconnect();
$server->stop();
$this->components->info("Stopping server on {$host}:{$port}");
});
}
/**
* Gracefully disconnect all connections.
*/
protected function gracefullyDisconnect(): void
{
$this->laravel->make(ApplicationProvider::class)
->all()
->each(function (Application $application) {
collect(
$this->laravel->make(ChannelManager::class)
->for($application)
->connections()
)->each->disconnect();
});
}
/**
* Schedule Pulse to ingest events if enabled.
*/
protected function ensurePulseEventsAreCollected(LoopInterface $loop, int $interval): void
{
if (! $this->laravel->bound(\Laravel\Pulse\Pulse::class)) {
return;
}
$loop->addPeriodicTimer($interval, function () {
$this->laravel->make(\Laravel\Pulse\Pulse::class)->ingest();
});
}
/**
* Schedule Telescope to store entries if enabled.
*/
protected function ensureTelescopeEntriesAreCollected(LoopInterface $loop, int $interval): void
{
if (! $this->laravel->bound(\Laravel\Telescope\Contracts\EntriesRepository::class)) {
return;
}
$loop->addPeriodicTimer($interval, function () {
\Laravel\Telescope\Telescope::store($this->laravel->make(\Laravel\Telescope\Contracts\EntriesRepository::class));
});
}
/**
* Get the list of signals handled by the command.
*/
public function getSubscribedSignals(): array
{
if (! windows_os()) {
return [SIGINT, SIGTERM, SIGTSTP];
}
$this->handleSignalWindows();
return [];
}
/**
* Handle the signals sent to the server.
*/
public function handleSignal(int $signal = 0, int|false $previousExitCode = 0): int|false
{
$this->components->info('Gracefully terminating connections.');
$this->gracefullyDisconnect();
return $previousExitCode;
}
/**
* Handle the signals sent to the server on Windows.
*/
public function handleSignalWindows(): void
{
if (function_exists('sapi_windows_set_ctrl_handler')) {
sapi_windows_set_ctrl_handler(fn () => exit($this->handleSignal()));
}
}
}
@@ -0,0 +1,21 @@
<?php
namespace Laravel\Reverb\Servers\Reverb\Contracts;
interface PubSubIncomingMessageHandler
{
/**
* Handle an incoming message from the PubSub provider.
*/
public function handle(string $payload): void;
/**
* Listen for the given event.
*/
public function listen(string $event, callable $callback): void;
/**
* Stop listening for the given event.
*/
public function stopListening(string $event): void;
}
@@ -0,0 +1,46 @@
<?php
namespace Laravel\Reverb\Servers\Reverb\Contracts;
use React\EventLoop\LoopInterface;
use React\Promise\PromiseInterface;
interface PubSubProvider
{
/**
* Connect to the publisher.
*/
public function connect(LoopInterface $loop): void;
/**
* Disconnect from the publisher.
*/
public function disconnect(): void;
/**
* Subscribe to the publisher.
*/
public function subscribe(): void;
/**
* Listen for the given event.
*/
public function on(string $event, callable $callback): void;
/**
* Listen for the given event.
*
* @alias on
*/
public function listen(string $event, callable $callback): void;
/**
* Stop listening for the given event.
*/
public function stopListening(string $event): void;
/**
* Publish a payload to the publisher.
*/
public function publish(array $payload): PromiseInterface;
}
+143
View File
@@ -0,0 +1,143 @@
<?php
namespace Laravel\Reverb\Servers\Reverb;
use InvalidArgumentException;
use Laravel\Reverb\Certificate;
use Laravel\Reverb\Contracts\ApplicationProvider;
use Laravel\Reverb\Protocols\Pusher\Contracts\ChannelConnectionManager;
use Laravel\Reverb\Protocols\Pusher\Contracts\ChannelManager;
use Laravel\Reverb\Protocols\Pusher\Http\Controllers\ChannelController;
use Laravel\Reverb\Protocols\Pusher\Http\Controllers\ChannelsController;
use Laravel\Reverb\Protocols\Pusher\Http\Controllers\ChannelUsersController;
use Laravel\Reverb\Protocols\Pusher\Http\Controllers\ConnectionsController;
use Laravel\Reverb\Protocols\Pusher\Http\Controllers\EventsBatchController;
use Laravel\Reverb\Protocols\Pusher\Http\Controllers\EventsController;
use Laravel\Reverb\Protocols\Pusher\Http\Controllers\HealthCheckController;
use Laravel\Reverb\Protocols\Pusher\Http\Controllers\PusherController;
use Laravel\Reverb\Protocols\Pusher\Http\Controllers\UsersTerminateController;
use Laravel\Reverb\Protocols\Pusher\Managers\ArrayChannelConnectionManager;
use Laravel\Reverb\Protocols\Pusher\Managers\ArrayChannelManager;
use Laravel\Reverb\Protocols\Pusher\PusherPubSubIncomingMessageHandler;
use Laravel\Reverb\Protocols\Pusher\Server as PusherServer;
use Laravel\Reverb\Servers\Reverb\Contracts\PubSubIncomingMessageHandler;
use Laravel\Reverb\Servers\Reverb\Http\Route;
use Laravel\Reverb\Servers\Reverb\Http\Router;
use Laravel\Reverb\Servers\Reverb\Http\Server as HttpServer;
use React\EventLoop\Loop;
use React\EventLoop\LoopInterface;
use React\Socket\SocketServer;
use Symfony\Component\Routing\Matcher\UrlMatcher;
use Symfony\Component\Routing\RequestContext;
use Symfony\Component\Routing\RouteCollection;
class Factory
{
/**
* Create a new WebSocket server instance.
*/
public static function make(
string $host = '0.0.0.0',
string $port = '8080',
string $path = '',
?string $hostname = null,
int $maxRequestSize = 10_000,
array $options = [],
string $protocol = 'pusher',
?LoopInterface $loop = null
): HttpServer {
$loop = $loop ?: Loop::get();
$router = match ($protocol) {
'pusher' => static::makePusherRouter($path),
default => throw new InvalidArgumentException("Unsupported protocol [{$protocol}]."),
};
$options['tls'] = static::configureTls($options['tls'] ?? [], $hostname);
$uri = static::usesTls($options['tls']) ? "tls://{$host}:{$port}" : "{$host}:{$port}";
return new HttpServer(
new SocketServer($uri, $options, $loop),
$router,
$maxRequestSize,
$loop
);
}
/**
* Create a new WebSocket server for the Pusher protocol.
*/
public static function makePusherRouter(string $path): Router
{
app()->singleton(
ChannelManager::class,
fn () => new ArrayChannelManager
);
app()->bind(
ChannelConnectionManager::class,
fn () => new ArrayChannelConnectionManager
);
app()->singleton(
PubSubIncomingMessageHandler::class,
fn () => new PusherPubSubIncomingMessageHandler,
);
return new Router(new UrlMatcher(static::pusherRoutes($path), new RequestContext));
}
/**
* Generate the routes required to handle Pusher requests.
*/
protected static function pusherRoutes(string $path): RouteCollection
{
$routes = new RouteCollection;
$routes->add('sockets', Route::get('/app/{appKey}', new PusherController(app(PusherServer::class), app(ApplicationProvider::class))));
$routes->add('events', Route::post('/apps/{appId}/events', new EventsController));
$routes->add('events_batch', Route::post('/apps/{appId}/batch_events', new EventsBatchController));
$routes->add('connections', Route::get('/apps/{appId}/connections', new ConnectionsController));
$routes->add('channels', Route::get('/apps/{appId}/channels', new ChannelsController));
$routes->add('channel', Route::get('/apps/{appId}/channels/{channel}', new ChannelController));
$routes->add('channel_users', Route::get('/apps/{appId}/channels/{channel}/users', new ChannelUsersController));
$routes->add('users_terminate', Route::post('/apps/{appId}/users/{userId}/terminate_connections', new UsersTerminateController));
$routes->add('health_check', Route::get('/up', new HealthCheckController));
$routes->addPrefix($path);
return $routes;
}
/**
* Configure the TLS context for the server.
*
* @param array $context<string, mixed>
* @return array<string, mixed>
*/
protected static function configureTls(array $context, ?string $hostname): array
{
$context = array_filter($context, fn ($value) => $value !== null);
if (! static::usesTls($context) && $hostname && Certificate::exists($hostname)) {
[$certificate, $key] = Certificate::resolve($hostname);
$context['local_cert'] = $certificate;
$context['local_pk'] = $key;
$context['verify_peer'] = app()->environment() === 'production';
}
return $context;
}
/**
* Determine whether the server uses TLS.
*
* @param array $context<string, mixed>
*/
protected static function usesTls(array $context): bool
{
return ($context['local_cert'] ?? false) || ($context['local_pk'] ?? false);
}
}
@@ -0,0 +1,128 @@
<?php
namespace Laravel\Reverb\Servers\Reverb\Http;
use BadMethodCallException;
use React\Socket\ConnectionInterface;
class Connection
{
/**
* Connection ID.
*/
protected int $id;
/**
* Connection status.
*/
protected bool $connected = false;
/**
* Connection buffer.
*/
protected string $buffer = '';
/**
* Create a new connection instance.
*/
public function __construct(protected ConnectionInterface $connection)
{
$this->id = (int) $connection->stream;
}
/**
* Return the connection ID.
*/
public function id(): int
{
return $this->id;
}
/**
* Mark the connection as connected.
*/
public function connect(): void
{
$this->connected = true;
}
/**
* Determine whether the connection is connected.
*/
public function isConnected(): bool
{
return $this->connected;
}
/**
* Get the HTTP message buffer.
*/
public function buffer(): string
{
return $this->buffer;
}
/**
* Determine whether the connection has an HTTP message buffer set.
*/
public function hasBuffer(): bool
{
return $this->buffer !== '';
}
/**
* Return the HTTP message buffer length.
*/
public function bufferLength(): int
{
return strlen($this->buffer);
}
/**
* Append to the HTTP message buffer.
*/
public function appendToBuffer($message): void
{
$this->buffer .= $message;
}
/**
* Clear the HTTP message buffer.
*/
public function clearBuffer(): void
{
$this->buffer = '';
}
/**
* Send a message to the connection.
*/
public function send($data): self
{
$this->connection->write($data);
return $this;
}
/**
* Close the connection.
*/
public function close(): self
{
$this->connection->end();
return $this;
}
/**
* Dynamically proxy method calls to the underlying connection.
*/
public function __call($method, $parameters)
{
if (! method_exists($this->connection, $method)) {
throw new BadMethodCallException("Method [{$method}] does not exist on [".get_class($this->connection).'].');
}
return $this->connection->{$method}(...$parameters);
}
}
@@ -0,0 +1,55 @@
<?php
namespace Laravel\Reverb\Servers\Reverb\Http;
use GuzzleHttp\Psr7\Message;
use OverflowException;
use Psr\Http\Message\RequestInterface;
class Request
{
/**
* End of message delimiter.
*
* @var string
*/
const EOM = "\r\n\r\n";
/**
* Turn the raw message into a Psr7 request.
*/
public static function from(string $message, Connection $connection, int $maxRequestSize): ?RequestInterface
{
$connection->appendToBuffer($message);
if ($connection->bufferLength() > $maxRequestSize) {
throw new OverflowException('Maximum HTTP buffer size of '.$maxRequestSize.'exceeded.');
}
if (static::isEndOfMessage($buffer = $connection->buffer())) {
$request = Message::parseRequest($buffer);
if (! $contentLength = $request->getHeader('Content-Length')) {
return $request;
}
if ($request->getBody()->getSize() < $contentLength[0] ?? 0) {
return null;
}
$connection->clearBuffer();
return $request;
}
return null;
}
/**
* Determine if the message has been buffered as per the HTTP specification
*/
protected static function isEndOfMessage(string $message): bool
{
return (bool) strpos($message, static::EOM);
}
}
@@ -0,0 +1,18 @@
<?php
namespace Laravel\Reverb\Servers\Reverb\Http;
use Symfony\Component\HttpFoundation\JsonResponse;
class Response extends JsonResponse
{
/**
* Create a new Http response instance.
*/
public function __construct(mixed $data = null, int $status = 200, array $headers = [], bool $json = false)
{
parent::__construct($data, $status, $headers, $json);
$this->headers->set('Content-Length', (string) strlen($this->content));
}
}
+104
View File
@@ -0,0 +1,104 @@
<?php
namespace Laravel\Reverb\Servers\Reverb\Http;
use Illuminate\Support\Arr;
use Symfony\Component\Routing\Loader\Configurator\Traits\RouteTrait;
use Symfony\Component\Routing\Route as BaseRoute;
class Route
{
use RouteTrait;
/**
* Create a new route instance.
*/
public function __construct(string $path)
{
$this->route = new BaseRoute($path);
}
/**
* Create a new `GET` route.
*/
public static function get(string $path, callable $action): BaseRoute
{
return static::route($path, 'GET', $action);
}
/**
* Create a new `POST` route.
*/
public static function post($path, callable $action): BaseRoute
{
return static::route($path, 'POST', $action);
}
/**
* Create a new `PUT` route.
*/
public static function put($path, callable $action): BaseRoute
{
return static::route($path, 'PUT', $action);
}
/**
* Create a new `PATCH` route.
*/
public static function patch($path, callable $action): BaseRoute
{
return static::route($path, 'PATCH', $action);
}
/**
* Create a new `DELETE` route.
*/
public static function delete($path, callable $action): BaseRoute
{
return static::route($path, 'DELETE', $action);
}
/**
* Create a new `HEAD` route.
*/
public static function head($path, callable $action): BaseRoute
{
return static::route($path, 'HEAD', $action);
}
/**
* Create a new `CONNECT` route.
*/
public static function connect($path, callable $action): BaseRoute
{
return static::route($path, 'CONNECT', $action);
}
/**
* Create a new `OPTIONS` route.
*/
public static function options($path, callable $action): BaseRoute
{
return static::route($path, 'OPTIONS', $action);
}
/**
* Create a new `TRACE` route.
*/
public static function trace($path, callable $action): BaseRoute
{
return static::route($path, 'TRACE', $action);
}
/**
* Create a new route.
*/
protected static function route(string $path, string|array $methods, callable $action): BaseRoute
{
$route = (new static($path))
->methods(Arr::wrap($methods))
->controller($action);
return $route->route;
}
}
+152
View File
@@ -0,0 +1,152 @@
<?php
namespace Laravel\Reverb\Servers\Reverb\Http;
use Closure;
use GuzzleHttp\Psr7\HttpFactory;
use GuzzleHttp\Psr7\Message;
use Illuminate\Support\Arr;
use Laravel\Reverb\Servers\Reverb\Concerns\ClosesConnections;
use Laravel\Reverb\Servers\Reverb\Connection as ReverbConnection;
use Psr\Http\Message\RequestInterface;
use Ratchet\RFC6455\Handshake\RequestVerifier;
use Ratchet\RFC6455\Handshake\ServerNegotiator;
use React\Promise\PromiseInterface;
use ReflectionFunction;
use ReflectionMethod;
use Symfony\Component\Routing\Exception\MethodNotAllowedException;
use Symfony\Component\Routing\Exception\ResourceNotFoundException;
use Symfony\Component\Routing\Matcher\UrlMatcherInterface;
class Router
{
use ClosesConnections;
/**
* The server negotiator instance.
*/
protected ServerNegotiator $negotiator;
/**
* Create a new router instance.
*/
public function __construct(protected UrlMatcherInterface $matcher)
{
$this->negotiator = new ServerNegotiator(new RequestVerifier, new HttpFactory);
}
/**
* Dispatch the request to the appropriate controller.
*/
public function dispatch(RequestInterface $request, Connection $connection): mixed
{
$uri = $request->getUri();
$context = $this->matcher->getContext();
$context->setMethod($request->getMethod());
$context->setHost($uri->getHost());
try {
$route = $this->matcher->match($uri->getPath());
} catch (MethodNotAllowedException $e) {
$this->close($connection, 405, 'Method not allowed.', ['Allow' => $e->getAllowedMethods()]);
return null;
} catch (ResourceNotFoundException $e) {
$this->close($connection, 404, 'Not found.');
return null;
}
$controller = $this->controller($route);
if ($this->isWebSocketRequest($request)) {
$wsConnection = $this->attemptUpgrade($request, $connection);
return $controller($request, $wsConnection, ...Arr::except($route, ['_controller', '_route']));
}
$routeParameters = Arr::except($route, [
'_controller',
'_route',
]) + ['request' => $request, 'connection' => $connection];
$response = $controller(
...$this->arguments($controller, $routeParameters)
);
return $response instanceof PromiseInterface ?
$response->then(fn ($response) => $connection->send($response)->close()) :
$connection->send($response)->close();
}
/**
* Get the controller callable for the given route.
*
* @param array<string, mixed> $route
*/
protected function controller(array $route): callable
{
return $route['_controller'];
}
/**
* Determine whether the request is for a WebSocket connection.
*/
protected function isWebSocketRequest(RequestInterface $request): bool
{
return $request->getHeader('Upgrade')[0] ?? null === 'websocket';
}
/**
* Negotiate the WebSocket connection upgrade.
*/
protected function attemptUpgrade(RequestInterface $request, Connection $connection): ReverbConnection
{
$response = $this->negotiator->handshake($request)
->withHeader('X-Powered-By', 'Laravel Reverb');
$connection->write(Message::toString($response));
return new ReverbConnection($connection);
}
/**
* Get the arguments for the controller.
*
* @return array<int, mixed>
*/
protected function arguments(callable $controller, array $routeParameters): array
{
$parameters = $this->parameters($controller);
return array_map(function ($parameter) use ($routeParameters) {
return $routeParameters[$parameter['name']] ?? null;
}, $parameters);
}
/**
* Get the parameters for the controller.
*
* @return array<int, array{ name: string, type: string, position: int }>
*/
protected function parameters(mixed $controller): array
{
$method = match (true) {
$controller instanceof Closure => new ReflectionFunction($controller),
is_string($controller) => count($parts = explode('::', $controller)) > 1 ? new ReflectionMethod(...$parts) : new ReflectionFunction($controller),
! is_array($controller) => new ReflectionMethod($controller, '__invoke'),
is_array($controller) => new ReflectionMethod($controller[0], $controller[1]),
};
$parameters = $method->getParameters();
return array_map(function ($parameter) {
return [
'name' => $parameter->getName(),
'type' => $parameter->getType()->getName(),
'position' => $parameter->getPosition(),
];
}, $parameters);
}
}
+118
View File
@@ -0,0 +1,118 @@
<?php
namespace Laravel\Reverb\Servers\Reverb\Http;
use Illuminate\Support\Str;
use Laravel\Reverb\Loggers\Log;
use Laravel\Reverb\Servers\Reverb\Concerns\ClosesConnections;
use OverflowException;
use Psr\Http\Message\RequestInterface;
use React\EventLoop\Loop;
use React\EventLoop\LoopInterface;
use React\Socket\ConnectionInterface;
use React\Socket\ServerInterface;
use Symfony\Component\HttpKernel\Exception\HttpException;
use Throwable;
class Server
{
use ClosesConnections;
/**
* Create a new Http server instance.
*/
public function __construct(protected ServerInterface $socket, protected Router $router, protected int $maxRequestSize, protected ?LoopInterface $loop = null)
{
gc_disable();
$this->loop = $loop ?: Loop::get();
$this->loop->addPeriodicTimer(30, fn () => gc_collect_cycles());
// Register __invoke handler for this class to receive new connections...
$socket->on('connection', $this);
}
/**
* Start the Http server
*/
public function start(): void
{
try {
$this->loop->run();
} catch (Throwable $e) {
Log::error($e->getMessage());
}
}
/**
* Handle an incoming request.
*/
protected function handleRequest(string $message, Connection $connection): void
{
if ($connection->isConnected()) {
return;
}
if (($request = $this->createRequest($message, $connection)) === null) {
return;
}
$connection->connect();
try {
$this->router->dispatch($request, $connection);
} catch (HttpException $e) {
$this->close($connection, $e->getStatusCode(), $e->getMessage());
} catch (Throwable $e) {
Log::error($e->getMessage());
$this->close($connection, 500, 'Internal server error.');
}
}
/**
* Create a Psr7 request from the incoming message.
*/
protected function createRequest(string $message, Connection $connection): ?RequestInterface
{
try {
$request = Request::from($message, $connection, $this->maxRequestSize);
} catch (OverflowException $e) {
$this->close($connection, 413, 'Payload too large.');
} catch (Throwable $e) {
$this->close($connection, 400, 'Bad request.');
}
return $request ?? null;
}
/**
* Stop the Http server
*/
public function stop(): void
{
$this->loop->stop();
$this->socket->close();
}
/**
* Invoke the server with a new connection instance.
*/
public function __invoke(ConnectionInterface $connection): void
{
$connection = new Connection($connection);
$connection->on('data', function ($data) use ($connection) {
$this->handleRequest($data, $connection);
});
}
/**
* Determine whether the server has TLS support.
*/
public function isSecure(): bool
{
return Str::startsWith($this->socket->getAddress(), 'tls://');
}
}
@@ -0,0 +1,219 @@
<?php
namespace Laravel\Reverb\Servers\Reverb\Publishing;
use Clue\React\Redis\Client;
use Exception;
use Illuminate\Support\Arr;
use Illuminate\Support\ConfigurationUrlParser;
use Illuminate\Support\Facades\Config;
use Laravel\Reverb\Exceptions\RedisConnectionException;
use Laravel\Reverb\Loggers\Log;
use React\EventLoop\LoopInterface;
class RedisClient
{
/**
* Redis connection client.
*
* @var \Clue\React\Redis\Client
*/
protected $client;
/**
* The name of the Redis connection.
*/
protected string $name = 'redis';
/**
* Determine if the client should attempt to reconnect when disconnected from the server.
*/
protected bool $shouldRetry = true;
/**
* Number of seconds the elapsed since attempting to reconnect.
*/
protected int $retryTimer = 0;
/**
* Create a new instance of the Redis client.
*
* @param callable|null $onConnect
*/
public function __construct(
protected LoopInterface $loop,
protected RedisClientFactory $clientFactory,
protected string $channel,
protected array $server,
protected $onConnect = null
) {
//
}
/**
* Create a new connetion to the Redis server.
*/
public function connect(): void
{
$this->clientFactory->make($this->loop, $this->redisUrl())->then(
fn (Client $client) => $this->onConnection($client),
fn (Exception $exception) => $this->onFailedConnection($exception),
);
}
/**
* Attempt to reconnect to the Redis server.
*/
public function reconnect(): void
{
if (! $this->shouldRetry) {
return;
}
$this->loop->addTimer(1, fn () => $this->attemptReconnection());
}
/**
* Disconnect from the Redis server.
*/
public function disconnect(): void
{
$this->shouldRetry = false;
$this->client?->close();
}
/**
* Listen for a given event.
*/
public function on(string $event, callable $callback): void
{
$this->client->on($event, $callback);
}
/**
* Determine if the client is currently connected to the server.
*/
public function isConnected(): bool
{
return (bool) $this->client === true && $this->client instanceof Client;
}
/**
* Handle a connection failure to the Redis server.
*/
protected function configureClientErrorHandler(): void
{
$this->client->on('close', function () {
$this->client = null;
Log::info('Disconnected from Redis', "<fg=red>{$this->name}</>");
$this->reconnect();
});
}
/**
* Handle a successful connection to the Redis server.
*/
protected function onConnection(Client $client): void
{
$this->client = $client;
$this->resetRetryTimer();
$this->configureClientErrorHandler();
if ($this->onConnect) {
call_user_func($this->onConnect, $client);
}
Log::info('Redis connection established', "<fg=green>{$this->name}</>");
}
/**
* Handle a failed connection to the Redis server.
*/
protected function onFailedConnection(Exception $exception): void
{
$this->client = null;
Log::error($exception->getMessage());
$this->reconnect();
}
/**
* Attempt to reconnect to the Redis server until the timeout is reached.
*/
protected function attemptReconnection(): void
{
$this->retryTimer++;
if ($this->retryTimer >= $this->retryTimeout()) {
$exception = RedisConnectionException::failedAfter($this->name, $this->retryTimeout());
Log::error($exception->getMessage());
throw $exception;
}
Log::info('Attempting reconnection to Redis', "<fg=yellow>{$this->name}</>");
$this->connect();
}
/**
* Determine the configured reconnection timeout.
*/
protected function retryTimeout(): int
{
return (int) ($this->server['timeout'] ?? 60);
}
/**
* Reset the retry connection timer.
*/
protected function resetRetryTimer(): void
{
$this->retryTimer = 0;
}
/**
* Get the connection URL for Redis.
*/
protected function redisUrl(): string
{
$config = empty($this->server) ? Config::get('database.redis.default') : $this->server;
$parsed = (new ConfigurationUrlParser)->parseConfiguration($config);
$driver = strtolower($parsed['driver'] ?? '');
if (in_array($driver, ['tcp', 'tls'])) {
$parsed['scheme'] = $driver;
}
[$host, $port, $protocol, $query] = [
$parsed['host'],
$parsed['port'] ?: 6379,
Arr::get($parsed, 'scheme') === 'tls' ? 's' : '',
[],
];
if ($parsed['username'] ?? false) {
$query['username'] = $parsed['username'];
}
if ($parsed['password'] ?? false) {
$query['password'] = $parsed['password'];
}
if ($parsed['database'] ?? false) {
$query['db'] = $parsed['database'];
}
$query = http_build_query($query);
return "redis{$protocol}://{$host}:{$port}".($query ? "?{$query}" : '');
}
}
@@ -0,0 +1,20 @@
<?php
namespace Laravel\Reverb\Servers\Reverb\Publishing;
use Clue\React\Redis\Factory;
use React\EventLoop\LoopInterface;
use React\Promise\PromiseInterface;
class RedisClientFactory
{
/**
* Create a new Redis client.
*/
public function make(LoopInterface $loop, string $redisUrl): PromiseInterface
{
return (new Factory($loop))->createClient(
$redisUrl
);
}
}
@@ -0,0 +1,106 @@
<?php
namespace Laravel\Reverb\Servers\Reverb\Publishing;
use Laravel\Reverb\Servers\Reverb\Contracts\PubSubIncomingMessageHandler;
use Laravel\Reverb\Servers\Reverb\Contracts\PubSubProvider;
use React\EventLoop\LoopInterface;
use React\Promise\PromiseInterface;
class RedisPubSubProvider implements PubSubProvider
{
/**
* The Redis publisher client.
*
* @var \Laravel\Reverb\Servers\Reverb\Publishing\RedisPublishClient
*/
protected $publisher;
/**
* The Redis subscriber client.
*
* @var \Laravel\Reverb\Servers\Reverb\Publishing\RedisSubscribeClient
*/
protected $subscriber;
/**
* Instantiate a new instance of the provider.
*/
public function __construct(
protected RedisClientFactory $clientFactory,
protected PubSubIncomingMessageHandler $messageHandler,
protected string $channel,
protected array $server = []
) {
//
}
/**
* Connect to the publisher.
*/
public function connect(LoopInterface $loop): void
{
$properties = [$loop, $this->clientFactory, $this->channel, $this->server];
$this->publisher = new RedisPublishClient(...$properties);
$this->subscriber = new RedisSubscribeClient(...array_merge($properties, [fn () => $this->subscribe()]));
$this->publisher->connect();
$this->subscriber->connect();
}
/**
* Disconnect from the publisher.
*/
public function disconnect(): void
{
$this->subscriber?->disconnect();
$this->publisher?->disconnect();
}
/**
* Subscribe to the publisher.
*/
public function subscribe(): void
{
$this->subscriber->subscribe();
$this->subscriber->on('message', function (string $channel, string $payload) {
$this->messageHandler->handle($payload);
});
}
/**
* Listen for a given event.
*/
public function on(string $event, callable $callback): void
{
$this->messageHandler->listen($event, $callback);
}
/**
* Listen for the given event.
*
* @alias on
*/
public function listen(string $event, callable $callback): void
{
$this->on($event, $callback);
}
/**
* Stop listening for the given event..
*/
public function stopListening(string $event): void
{
$this->messageHandler->stopListening($event);
}
/**
* Publish a payload to the publishingClientReconnectionTimer.
*/
public function publish(array $payload): PromiseInterface
{
return $this->publisher->publish($payload);
}
}
@@ -0,0 +1,65 @@
<?php
namespace Laravel\Reverb\Servers\Reverb\Publishing;
use Clue\React\Redis\Client;
use React\Promise\Promise;
use React\Promise\PromiseInterface;
use RuntimeException;
class RedisPublishClient extends RedisClient
{
/**
* The name of the Redis connection.
*/
protected string $name = 'publisher';
/**
* Stream of events queued while disconnected from Redis.
*/
protected $queuedEvents = [];
/**
* Queue the given publish event.
*/
protected function queueEvent(array $payload): void
{
$this->queuedEvents[] = $payload;
}
/**
* Process the queued events.
*/
protected function processQueuedEvents(): void
{
foreach ($this->queuedEvents as $event) {
$this->publish($event);
}
$this->queuedEvents = [];
}
/**
* Publish an event to the given channel.
*/
public function publish(array $payload): PromiseInterface
{
if (! $this->isConnected($this->client)) {
$this->queueEvent($payload);
return new Promise(fn () => new RuntimeException);
}
return $this->client->publish($this->channel, json_encode($payload));
}
/**
* Handle a successful connection to the Redis server.
*/
protected function onConnection(Client $client): void
{
parent::onConnection($client);
$this->processQueuedEvents();
}
}
@@ -0,0 +1,19 @@
<?php
namespace Laravel\Reverb\Servers\Reverb\Publishing;
class RedisSubscribeClient extends RedisClient
{
/**
* The name of the Redis connection.
*/
protected string $name = 'subscriber';
/**
* Subscribe to the given Redis channel.
*/
public function subscribe(): void
{
$this->client->subscribe($this->channel);
}
}
@@ -0,0 +1,20 @@
<?php
namespace Laravel\Reverb\Servers\Reverb;
use Clue\React\Redis\Client;
use Clue\React\Redis\Factory;
use React\EventLoop\LoopInterface;
class RedisClientFactory
{
/**
* Create a new Redis client.
*/
public function make(LoopInterface $loop, string $redisUrl): Client
{
return (new Factory($loop))->createLazyClient(
$redisUrl
);
}
}
@@ -0,0 +1,75 @@
<?php
namespace Laravel\Reverb\Servers\Reverb;
use Illuminate\Console\Application as Artisan;
use Illuminate\Contracts\Foundation\Application;
use Laravel\Reverb\Contracts\ServerProvider;
use Laravel\Reverb\Servers\Reverb\Console\Commands\RestartServer;
use Laravel\Reverb\Servers\Reverb\Console\Commands\StartServer;
use Laravel\Reverb\Servers\Reverb\Contracts\PubSubIncomingMessageHandler;
use Laravel\Reverb\Servers\Reverb\Contracts\PubSubProvider;
use Laravel\Reverb\Servers\Reverb\Publishing\RedisClientFactory;
use Laravel\Reverb\Servers\Reverb\Publishing\RedisPubSubProvider;
class ReverbServerProvider extends ServerProvider
{
/**
* Indicates whether the Reverb server should publish events.
*
* @var bool
*/
protected $publishesEvents;
/**
* Create a new Reverb server provider instance.
*/
public function __construct(protected Application $app, protected array $config)
{
$this->publishesEvents = (bool) $this->config['scaling']['enabled'] ?? false;
}
/**
* Register any application services.
*/
public function register(): void
{
$this->app->singleton(PubSubProvider::class, fn ($app) => new RedisPubSubProvider(
$app->make(RedisClientFactory::class),
$app->make(PubSubIncomingMessageHandler::class),
$this->config['scaling']['channel'] ?? 'reverb',
$this->config['scaling']['server'] ?? []
));
}
/**
* Bootstrap any application services.
*/
public function boot(): void
{
if ($this->app->runningInConsole()) {
Artisan::starting(function ($artisan) {
$artisan->resolveCommands([
StartServer::class,
RestartServer::class,
]);
});
}
}
/**
* Enable publishing of events.
*/
public function withPublishing(): void
{
$this->publishesEvents = true;
}
/**
* Determine whether the server should publish events.
*/
public function shouldPublishEvents(): bool
{
return $this->publishesEvents;
}
}