Adds Pusher routes (#16)

* close with message

* ensure app exists

* handle event trigger

* scaffold routes

* implement events controller

* update phpunit config

* implement batch events endpoint

* implement channels route

* wip

* add channels test

* use correct id

* wip

* revert

* add channel endpoint

* Fix code styling

* implements users routes

* implement terminate user endpoint

* formatting

composer.json

@@ -34,7 +34,8 @@
         "pestphp/pest": "^2.0",
         "phpstan/phpstan": "^1.10",
         "ratchet/pawl": "^0.4.1",
-        "react/async": "^4.0"
+        "react/async": "^4.0",
+        "react/http": "^1.9"
     },
     "autoload": {
         "psr-4": {

phpunit.xml.dist

@@ -1,23 +1,24 @@
 <?xml version="1.0" encoding="UTF-8"?>
-<phpunit xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" colors="true" xsi:noNamespaceSchemaLocation="https://schema.phpunit.de/10.1/phpunit.xsd">
-    <testsuites>
-        <testsuite name="Unit">
-            <directory suffix="Test.php">./tests/Unit</directory>
-        </testsuite>
-        <testsuite name="Feature">
-            <directory suffix="Test.php">./tests/Feature</directory>
-        </testsuite>
-    </testsuites>
-    <coverage>
-        <include>
-            <directory suffix=".php">./src</directory>
-        </include>
-    </coverage>
-    <php>
-        <env name="APP_KEY" value="base64:uz4B1RtFO57QGzbZX1kRYX9hIRB50+QzqFeg9zbFJlY="/>
-        <env name="PUSHER_APP_ID" value="123456"/>
-        <env name="PUSHER_APP_KEY" value="pusher-key"/>
-        <env name="PUSHER_APP_SECRET" value="pusher-secret"/>
-        <env name="REVERB_API_GATEWAY_CONNECTION_CACHE" value="redis"/>
-    </php>
-</phpunit>
+<phpunit xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" colors="true" xsi:noNamespaceSchemaLocation="https://schema.phpunit.de/10.4/phpunit.xsd">
+  <testsuites>
+    <testsuite name="Unit">
+      <directory suffix="Test.php">./tests/Unit</directory>
+    </testsuite>
+    <testsuite name="Feature">
+      <directory suffix="Test.php">./tests/Feature</directory>
+    </testsuite>
+  </testsuites>
+  <coverage/>
+  <php>
+    <env name="APP_KEY" value="base64:uz4B1RtFO57QGzbZX1kRYX9hIRB50+QzqFeg9zbFJlY="/>
+    <env name="PUSHER_APP_ID" value="123456"/>
+    <env name="PUSHER_APP_KEY" value="pusher-key"/>
+    <env name="PUSHER_APP_SECRET" value="pusher-secret"/>
+    <env name="REVERB_API_GATEWAY_CONNECTION_CACHE" value="redis"/>
+  </php>
+  <source>
+    <include>
+      <directory suffix=".php">./src</directory>
+    </include>
+  </source>
+</phpunit>

src/Channels/Channel.php

@@ -51,11 +51,11 @@ public function broadcast(Application $app, array $payload, Connection $except =
     {
         collect(App::make(ChannelManager::class)->for($app)->connections($this))
             ->each(function ($connection) use ($payload, $except) {
-                if ($except && $except->identifier() === $connection->identifier()) {
+                if ($except && $except->id() === $connection->id()) {
                     return;
                 }
 
-                if (isset($payload['except']) && $payload['except'] === $connection->identifier()) {
+                if (isset($payload['except']) && $payload['except'] === $connection->id()) {
                     return;
                 }
 

src/ClientEvent.php

@@ -33,7 +33,7 @@ public static function whisper(Connection $connection, array $payload): void
     {
         Event::dispatch(
             $connection->app(),
-            $payload + ['except' => $connection->identifier()],
+            $payload + ['except' => $connection->id()],
             $connection
         );
     }

src/Concerns/ClosesConnections.php

@@ -11,9 +11,9 @@ trait ClosesConnections
     /**
      * Close the connection.
      */
-    protected function close(Connection $connection, int $statusCode = 400, array $headers = []): void
+    protected function close(Connection $connection, int $statusCode = 400, string $message = '', array $headers = []): void
     {
-        $response = new Response($statusCode, $headers);
+        $response = new Response($statusCode, $headers, $message);
 
         $connection->send(Message::toString($response));
         $connection->close();

src/Contracts/Connection.php

@@ -97,7 +97,7 @@ public function disconnect(): void
 
         App::make(ConnectionManager::class)
             ->for($this->app())
-            ->disconnect($this->identifier());
+            ->disconnect($this->id());
 
         $this->terminate();
     }

src/Event.php

@@ -2,6 +2,7 @@
 
 namespace Laravel\Reverb;
 
+use Illuminate\Support\Arr;
 use Illuminate\Support\Facades\App;
 use Laravel\Reverb\Channels\ChannelBroker;
 use Laravel\Reverb\Contracts\Connection;
@@ -33,10 +34,12 @@ public static function dispatch(Application $app, array $payload, Connection $co
      */
     public static function dispatchSynchronously(Application $app, array $payload, Connection $connection = null): void
     {
-        $channels = isset($payload['channel']) ? [$payload['channel']] : $payload['channels'];
+        $channels = Arr::wrap($payload['channels'] ?? $payload['channel'] ?? []);
 
         foreach ($channels as $channel) {
+            unset($payload['channels']);
             $channel = ChannelBroker::create($channel);
+            $payload['channel'] = $channel->name();
 
             $channel->broadcast($app, $payload, $connection);
         }

src/Http/Router.php

@@ -38,12 +38,18 @@ public function dispatch(RequestInterface $request, Connection $connection): mix
         try {
             $route = $this->matcher->match($uri->getPath());
         } catch (MethodNotAllowedException $e) {
-            return $this->close($connection, 405, ['Allow' => $e->getAllowedMethods()]);
+            return $this->close($connection, 405, 'Method now allowed', ['Allow' => $e->getAllowedMethods()]);
         } catch (ResourceNotFoundException $e) {
-            return $this->close($connection, 404);
+            return $this->close($connection, 404, 'Not found.');
         }
 
-        return $route['_controller']($request, $connection, ...Arr::except($route, ['_controller', '_route']));
+        $response = $route['_controller']($request, $connection, ...Arr::except($route, ['_controller', '_route']));
+
+        if (! $this->isWebSocketRequest($request)) {
+            return $connection->send($response)->close();
+        }
+
+        return null;
     }
 
     /**

src/Http/Server.php

@@ -45,6 +45,15 @@ public function start(): void
         $this->loop->run();
     }
 
+    /**
+     * Stop the Http server
+     */
+    public function stop(): void
+    {
+        $this->loop->stop();
+        $this->socket->close();
+    }
+
     /**
      * Handle an incoming request.
      */
@@ -71,7 +80,7 @@ protected function createRequest(string $message, Connection $connection): Reque
         try {
             return Request::from($message, $connection);
         } catch (OverflowException $e) {
-            $this->close($connection, 413);
+            $this->close($connection, 413, 'Payload too large.');
         }
     }
 }

src/Managers/ChannelManager.php

@@ -45,15 +45,15 @@ public function app(): ?Application
      */
     public function subscribe(Channel $channel, Connection $connection, $data = []): void
     {
-        $this->connections[$this->application->id()][$channel->name()][$connection->identifier()] = $connection;
+        $this->connections[$this->application->id()][$channel->name()][$connection->id()] = $connection;
     }
 
     /**
      * Unsubscribe from a channel.
      */
     public function unsubscribe(Channel $channel, Connection $connection): void
     {
-        unset($this->connections[$this->application->id()][$channel->name()][$connection->identifier()]);
+        unset($this->connections[$this->application->id()][$channel->name()][$connection->id()]);
     }
 
     /**
@@ -89,16 +89,20 @@ public function connections(Channel $channel): array
     /**
      * Get the given channel from the cache.
      */
-    protected function channel(Channel $channel): Collection
+    public function channel(Channel $channel): Collection
     {
         return $this->channels($channel);
     }
 
     /**
      * Get the channels from the cache.
      */
-    protected function channels(Channel $channel = null): Collection
+    public function channels(Channel $channel = null): Collection
     {
+        if (! isset($this->connections[$this->application->id()])) {
+            $this->connections[$this->application->id()] = [];
+        }
+
         $channels = $this->connections[$this->application->id()];
 
         if ($channel) {

src/Managers/ConnectionManager.php

@@ -104,7 +104,7 @@ public function all(): array
      */
     public function save(Connection $connection): void
     {
-        $this->connections[$this->application->id()][$connection->identifier()] = $connection;
+        $this->connections[$this->application->id()][$connection->id()] = $connection;
     }
 
     /**

src/Pusher/Http/Controllers/ChannelController.php

@@ -0,0 +1,29 @@
+<?php
+
+namespace Laravel\Reverb\Pusher\Http\Controllers;
+
+use Laravel\Reverb\Channels\ChannelBroker;
+use Laravel\Reverb\Http\Connection;
+use Psr\Http\Message\RequestInterface;
+use Symfony\Component\HttpFoundation\JsonResponse;
+use Symfony\Component\HttpFoundation\Response;
+
+class ChannelController extends Controller
+{
+    /**
+     * Handle the request.
+     */
+    public function handle(RequestInterface $request, Connection $connection, ...$args): Response
+    {
+        $info = explode(',', $this->query['info'] ?? '');
+        $connections = $this->channels->channel(ChannelBroker::create($args['channel']));
+        $totalConnections = count($connections);
+
+        return new JsonResponse((object) array_filter([
+            'occupied' => $totalConnections > 0,
+            'user_count' => in_array('user_count', $info) ? $totalConnections : null,
+            'subscription_count' => in_array('subscription_count', $info) ? $totalConnections : null,
+            'cache' => in_array('cache', $info) ? '{}' : null,
+        ], fn ($item) => $item !== null));
+    }
+}

src/Pusher/Http/Controllers/ChannelUsersController.php

@@ -0,0 +1,27 @@
+<?php
+
+namespace Laravel\Reverb\Pusher\Http\Controllers;
+
+use Laravel\Reverb\Channels\ChannelBroker;
+use Laravel\Reverb\Channels\PresenceChannel;
+use Laravel\Reverb\Http\Connection;
+use Psr\Http\Message\RequestInterface;
+use Symfony\Component\HttpFoundation\JsonResponse;
+use Symfony\Component\HttpFoundation\Response;
+
+class ChannelUsersController extends Controller
+{
+    /**
+     * Handle the request.
+     */
+    public function handle(RequestInterface $request, Connection $connection, ...$args): Response
+    {
+        $channel = ChannelBroker::create($args['channel']);
+
+        if (! $channel instanceof PresenceChannel) {
+            return new JsonResponse((object) [], 400);
+        }
+
+        return new JsonResponse((object) []);
+    }
+}

src/Pusher/Http/Controllers/ChannelsController.php

@@ -0,0 +1,31 @@
+<?php
+
+namespace Laravel\Reverb\Pusher\Http\Controllers;
+
+use Illuminate\Support\Str;
+use Laravel\Reverb\Http\Connection;
+use Psr\Http\Message\RequestInterface;
+use Symfony\Component\HttpFoundation\JsonResponse;
+use Symfony\Component\HttpFoundation\Response;
+
+class ChannelsController extends Controller
+{
+    /**
+     * Handle the request.
+     */
+    public function handle(RequestInterface $request, Connection $connection, ...$args): Response
+    {
+        $channels = $this->channels->channels();
+        $info = explode(',', $this->query['info'] ?? '');
+
+        if (isset($this->query['filter_by_prefix'])) {
+            $channels = $channels->filter(fn ($connections, $name) => Str::startsWith($name, $this->query['filter_by_prefix']));
+        }
+
+        $channels = $channels->mapWithKeys(function ($connections, $name) use ($info) {
+            return [$name => array_filter(['user_count' => in_array('user_count', $info) ? count($connections) : null])];
+        });
+
+        return new JsonResponse((object) ['channels' => $channels]);
+    }
+}