Cleans up channels (#30)

* clean up channels

* Fix code styling

* revert

src/Channels/CacheChannel.php

@@ -14,7 +14,7 @@ class CacheChannel extends Channel
     /**
      * Send a message to all connections subscribed to the channel.
      */
-    public function broadcast(array $payload, Connection $except = null): void
+    public function broadcast(array $payload, ?Connection $except = null): void
     {
         $this->payload = $payload;
 
@@ -24,7 +24,7 @@ public function broadcast(array $payload, Connection $except = null): void
     /**
      * Broadcast a message triggered from an internal source.
      */
-    public function broadcastInternally(array $payload, Connection $except = null): void
+    public function broadcastInternally(array $payload, ?Connection $except = null): void
     {
         parent::broadcast($payload, $except);
     }

src/Channels/Channel.php

@@ -4,6 +4,7 @@
 
 use Laravel\Reverb\Concerns\SerializesChannels;
 use Laravel\Reverb\Contracts\ChannelConnectionManager;
+use Laravel\Reverb\Contracts\ChannelManager;
 use Laravel\Reverb\Contracts\Connection;
 
 class Channel
@@ -57,7 +58,7 @@ public function findById(string $id): ?Connection
     /**
      * Subscribe to the given channel.
      */
-    public function subscribe(Connection $connection, string $auth = null, string $data = null): void
+    public function subscribe(Connection $connection, ?string $auth = null, ?string $data = null): void
     {
         $this->connections->add($connection, $data ? json_decode($data, true) : []);
     }
@@ -68,6 +69,10 @@ public function subscribe(Connection $connection, string $auth = null, string $d
     public function unsubscribe(Connection $connection): void
     {
         $this->connections->remove($connection);
+
+        if ($this->connections->isEmpty()) {
+            app(ChannelManager::class)->for($connection->app())->remove($this);
+        }
     }
 
     /**
@@ -81,7 +86,7 @@ public function subscribed(Connection $connection): bool
     /**
      * Send a message to all connections subscribed to the channel.
      */
-    public function broadcast(array $payload, Connection $except = null): void
+    public function broadcast(array $payload, ?Connection $except = null): void
     {
         if ($except === null) {
             $this->broadcastToAll($payload);
@@ -115,7 +120,7 @@ public function broadcastToAll(array $payload): void
     /**
      * Broadcast a message triggered from an internal source.
      */
-    public function broadcastInternally(array $payload, Connection $except = null): void
+    public function broadcastInternally(array $payload, ?Connection $except = null): void
     {
         $this->broadcast($payload, $except);
     }

src/Channels/Concerns/InteractsWithPresenceChannels.php

@@ -11,7 +11,7 @@ trait InteractsWithPresenceChannels
     /**
      * Subscribe to the given channel.
      */
-    public function subscribe(Connection $connection, string $auth = null, string $data = null): void
+    public function subscribe(Connection $connection, ?string $auth = null, ?string $data = null): void
     {
         $this->verify($connection, $auth, $data);
 

src/Channels/Concerns/InteractsWithPrivateChannels.php

@@ -11,7 +11,7 @@ trait InteractsWithPrivateChannels
     /**
      * Subscribe to the given channel.
      */
-    public function subscribe(Connection $connection, string $auth = null, string $data = null): void
+    public function subscribe(Connection $connection, ?string $auth = null, ?string $data = null): void
     {
         $this->verify($connection, $auth, $data);
 
@@ -21,7 +21,7 @@ public function subscribe(Connection $connection, string $auth = null, string $d
     /**
      * Deteremine whether the given auth token is valid.
      */
-    protected function verify(Connection $connection, string $auth, string $data = null): bool
+    protected function verify(Connection $connection, string $auth, ?string $data = null): bool
     {
         $signature = "{$connection->id()}:{$this->name()}";
 

src/Contracts/ChannelConnectionManager.php

@@ -38,6 +38,11 @@ public function findById(string $id): ?ChannelConnection;
      */
     public function all(): array;
 
+    /**
+     * Determine whether any connections remain on the channel.
+     */
+    public function isEmpty(): bool;
+
     /**
      * Flush the channel connection manager.
      */

src/Contracts/ChannelManager.php

@@ -30,13 +30,18 @@ public function find(string $channel): Channel;
     /**
      * Get all the connections for the given channels.
      */
-    public function connections(string $channel = null): array;
+    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.
      */

src/Contracts/Logger.php

@@ -7,7 +7,7 @@ interface Logger
     /**
      * Log an infomational message.
      */
-    public function info(string $title, string $message = null): void;
+    public function info(string $title, ?string $message = null): void;
 
     /**
      * Log an error message.

src/Event.php

@@ -11,7 +11,7 @@ class Event
     /**
      * Dispatch a message to a channel.
      */
-    public static function dispatch(Application $app, array $payload, Connection $connection = null): void
+    public static function dispatch(Application $app, array $payload, ?Connection $connection = null): void
     {
         $server = app(ServerManager::class);
 
@@ -30,7 +30,7 @@ public static function dispatch(Application $app, array $payload, Connection $co
     /**
      * Notify all connections subscribed to the given channel.
      */
-    public static function dispatchSynchronously(Application $app, array $payload, Connection $connection = null): void
+    public static function dispatchSynchronously(Application $app, array $payload, ?Connection $connection = null): void
     {
         $channels = Arr::wrap($payload['channels'] ?? $payload['channel'] ?? []);
 

src/Loggers/CliLogger.php

@@ -24,7 +24,7 @@ public function __construct(protected OutputStyle $output)
     /**
      * Log an infomational message.
      */
-    public function info(string $title, string $message = null): void
+    public function info(string $title, ?string $message = null): void
     {
         $this->components->twoColumnDetail($title, $message);
     }

src/Loggers/NullLogger.php

@@ -9,7 +9,7 @@ class NullLogger implements Logger
     /**
      * Log an infomational message.
      */
-    public function info(string $title, string $message = null): void
+    public function info(string $title, ?string $message = null): void
     {
         //
     }

src/Loggers/StandardLogger.php

@@ -10,7 +10,7 @@ class StandardLogger implements Logger
     /**
      * Log an infomational message
      */
-    public function info(string $title, string $message = null): void
+    public function info(string $title, ?string $message = null): void
     {
         $output = $title;
 

src/Managers/ArrayChannelConnectionManager.php

@@ -67,6 +67,14 @@ 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.
      */

src/Managers/ArrayChannelManager.php

@@ -58,7 +58,7 @@ public function find(string $channel): Channel
      *
      * @return array<string, \Laravel\Reverb\Servers\Reverb\ChannelConnection>
      */
-    public function connections(string $channel = null): array
+    public function connections(?string $channel = null): array
     {
         $channels = Arr::wrap($this->channels($channel));
 
@@ -77,6 +77,14 @@ public function unsubscribeFromAll(Connection $connection): void
         }
     }
 
+    /**
+     * Remove the given channel.
+     */
+    public function remove(Channel $channel): void
+    {
+        unset($this->applications[$this->application->id()][$channel->name()]);
+    }
+
     /**
      * Get the given channel.
      */
@@ -90,7 +98,7 @@ public function channel(string $channel): Channel
      *
      * @return \Laravel\Reverb\Channels\Channel|array<string, \Laravel\Reverb\Channels\Channel>
      */
-    public function channels(string $channel = null): Channel|array
+    public function channels(?string $channel = null): Channel|array
     {
         if (! isset($this->applications[$this->application->id()])) {
             $this->applications[$this->application->id()] = [];

src/Managers/CacheChannelConnectionManager.php

@@ -87,6 +87,14 @@ public function findById(string $id): ?ChannelConnection
         );
     }
 
+    /**
+     * Determine whether any connections remain on the channel.
+     */
+    public function isEmpty(): bool
+    {
+        return empty($this->repository->get($this->key(), []));
+    }
+
     /**
      * Get all the connections.
      */

src/Managers/CacheChannelManager.php

@@ -58,7 +58,7 @@ public function find(string $channel): Channel
      *
      * @return array<string, \Laravel\Reverb\Servers\Reverb\ChannelConnection>
      */
-    public function connections(string $channel = null): array
+    public function connections(?string $channel = null): array
     {
         $channels = Arr::wrap($this->channels($channel));
 
@@ -77,6 +77,18 @@ public function unsubscribeFromAll(Connection $connection): void
         }
     }
 
+    /**
+     * Remove the given channel.
+     */
+    public function remove(Channel $channel): void
+    {
+        $channels = $this->channels();
+
+        unset($channels[$channel->name()]);
+
+        $this->repository->forever($this->prefix, $channels);
+    }
+
     /**
      * Get the given channel.
      */
@@ -90,7 +102,7 @@ public function channel(string $channel): Channel
      *
      * @return \Laravel\Reverb\Channels\Channel|array<string, \Laravel\Reverb\Channels\Channel>
      */
-    public function channels(string $channel = null): Channel|array
+    public function channels(?string $channel = null): Channel|array
     {
         $channels = $this->repository->get($this->prefix, []);
 

src/Pusher/Event.php

@@ -50,7 +50,7 @@ public function acknowledge(Connection $connection): void
     /**
      * Subscribe to the given channel.
      */
-    public function subscribe(Connection $connection, string $channel, string $auth = null, string $data = null): void
+    public function subscribe(Connection $connection, string $channel, ?string $auth = null, ?string $data = null): void
     {
         $channel = $this->channels
             ->for($connection->app())
@@ -122,7 +122,7 @@ public function ping(Connection $connection): void
     /**
      * Send a response to the given connection.
      */
-    public function send(Connection $connection, string $event, array $data = [], string $channel = null): void
+    public function send(Connection $connection, string $event, array $data = [], ?string $channel = null): void
     {
         $connection->send(
             static::formatPayload($event, $data, $channel)
@@ -132,7 +132,7 @@ public function send(Connection $connection, string $event, array $data = [], st
     /**
      * Send an internal response to the given connection.
      */
-    public function sendInternally(Connection $connection, string $event, array $data = [], string $channel = null): void
+    public function sendInternally(Connection $connection, string $event, array $data = [], ?string $channel = null): void
     {
         $connection->send(
             static::formatInternalPayload($event, $data, $channel)
@@ -142,7 +142,7 @@ public function sendInternally(Connection $connection, string $event, array $dat
     /**
      * Format the payload for the given event.
      */
-    public function formatPayload(string $event, array $data = [], string $channel = null, string $prefix = 'pusher:'): string|false
+    public function formatPayload(string $event, array $data = [], ?string $channel = null, string $prefix = 'pusher:'): string|false
     {
         return json_encode(
             array_filter([

src/Servers/Reverb/ChannelConnection.php

@@ -23,7 +23,7 @@ public function connection(): Connection
     /**
      * Get the connection data.
      */
-    public function data(string $key = null): mixed
+    public function data(?string $key = null): mixed
     {
         if ($key) {
             return Arr::get($this->data, $key);

src/Servers/Reverb/Connection.php

@@ -21,7 +21,7 @@ class Connection extends ConnectionContract
      */
     protected $hasBeenPinged = false;
 
-    public function __construct(protected WsConnection $connection, Application $application, string $origin = null)
+    public function __construct(protected WsConnection $connection, Application $application, ?string $origin = null)
     {
         parent::__construct($application, $origin);
     }

src/Servers/Reverb/Factory.php

@@ -25,7 +25,7 @@ class Factory
     /**
      * Create a new WebSocket server instance.
      */
-    public static function make(string $host = '0.0.0.0', string $port = '8080', LoopInterface $loop = null)
+    public static function make(string $host = '0.0.0.0', string $port = '8080', ?LoopInterface $loop = null)
     {
         $loop = $loop ?: Loop::get();
         $socket = new SocketServer("{$host}:{$port}", [], $loop);

src/WebSockets/WsConnection.php

@@ -113,7 +113,7 @@ public function withMaxMessageSize(int $size): void
     /**
      * Close the connection.
      */
-    public function close(FrameInterface $frame = null): void
+    public function close(?FrameInterface $frame = null): void
     {
         if ($frame) {
             $this->send($frame);

tests/ApiGatewayTestCase.php

@@ -118,7 +118,7 @@ public function send(array $message, ?string $connectionId = 'abc-123', $appKey
      *
      * @param  string  $appKey
      */
-    public function subscribe(string $channel, ?array $data = [], string $auth = null, ?string $connectionId = 'abc-123', $appKey = 'pusher-key'): void
+    public function subscribe(string $channel, ?array $data = [], ?string $auth = null, ?string $connectionId = 'abc-123', $appKey = 'pusher-key'): void
     {
         $data = ! empty($data) ? json_encode($data) : null;
 
@@ -161,7 +161,7 @@ public function disconnect($connectionId = 'abc-123'): void
      *
      * @return void
      */
-    public function assertSent(string $connectionId = null, mixed $message = null, int $times = null)
+    public function assertSent(?string $connectionId = null, mixed $message = null, ?int $times = null)
     {
         Bus::assertDispatched(SendToConnection::class, function ($job) use ($connectionId, $message) {
             return ($connectionId ? $job->connectionId === $connectionId : true)

tests/Connection.php

@@ -18,7 +18,7 @@ class Connection extends BaseConnection
 
     public $id;
 
-    public function __construct(string $identifier = null)
+    public function __construct(?string $identifier = null)
     {
         if ($identifier) {
             $this->identifier = $identifier;

tests/Feature/Reverb/ServerTest.php

@@ -422,6 +422,17 @@
     expect($message)->toBe('Maximum message size exceeded');
 });
 
+it('removes a channel when no subscribers remain', function () {
+    $connection = $this->connect();
+    $this->subscribe('test-channel', connection: $connection);
+
+    expect(channelManager()->all())->toHaveCount(1);
+
+    $this->unsubscribe('test-channel', $connection);
+
+    expect(channelManager()->all())->toHaveCount(0);
+});
+
 it('clears application state between requests', function () {
     $this->subscribe('test-channel');