Cleanup, configuration

src/Configuration.php

@@ -8,6 +8,8 @@ class Configuration
 
     private ?string $jwtSecret = null;
 
+    private bool $allowAnonymousSubscribe = false;
+
     public static function createDefault(): Configuration
     {
         return new Configuration();
@@ -34,5 +36,16 @@ class Configuration
     {
         return $this->jwtSecret;
     }
+
+	function getAllowAnonymousSubscribe():bool
+    {
+		return $this->allowAnonymousSubscribe;
+	}
+
+	function setAllowAnonymousSubscribe(bool $allowAnonymousSubscribe): self
+    {
+		$this->allowAnonymousSubscribe = $allowAnonymousSubscribe;
+		return $this;
+    }
 }
 

src/Daemon.php

@@ -23,7 +23,7 @@ class Daemon
 
     public function start(): void
     {
-        $this->server = new Server($this->config, []);
+        $this->server = new Server($this->config, $this->loop);
 
         $socket = new SocketServer("tcp://0.0.0.0:9000");
         $this->server->listen($socket);

src/Http/Middleware/MercureHandler.php

@@ -17,7 +17,6 @@ use React\Http\Message\Response;
 use React\Promise\Promise;
 use React\Promise\PromiseInterface;
 use React\Stream\ThroughStream;
-use SplObjectStorage;
 use Symfony\Component\Uid\Uuid;
 
 class MercureHandler
@@ -26,7 +25,6 @@ class MercureHandler
 
     public function __construct(
         private Configuration $config,
-        private SplObjectStorage $eventClients,
         private TopicManager $topicManager,
         ?LoopInterface $loop=null
     )
@@ -102,17 +100,25 @@ class MercureHandler
             $claims = $tok->claims->getAll();
             if (isset($claims['mercure']['subscribe'])) {
                 $subscribeClaims = $claims['mercure']['subscribe'];
-                // TODO check topic against publishClaims
                 if (!$this->checkTopicClaims($topics, $subscribeClaims)) {
-                    throw new SecurityException("Insufficient permissions for subscribe", SecurityException::ERR_NO_PERMISSION);
+                    throw new SecurityException(
+                        message: "Insufficient permissions for subscribe", 
+                        code: SecurityException::ERR_NO_PERMISSION
+                    );
                 }
             }
         } else {
-            // TODO add option to allow/disallow anonymous acess. should still respect 
+            // Disallow if we don't allow anonymous subscribers. Note that anonymous
+            // subscribers will not receive updates marked as private.
+            if (!$this->config->getAllowAnonymousSubscribe()) {
+                throw new SecurityException(
+                    message: "Anonymous access disallowed",
+                    code: SecurityException::ERR_ACCESS_DENIED
+                );
+            }
         }
         
         $this->topicManager->subscribe($subscriber, $topics);
-        //$this->eventClients->attach($responseStream, $request);
         $responseStream->on('close', function () use ($subscriber, $topics) {
             $this->topicManager->unsubscribe($subscriber, $topics);
         });
@@ -139,7 +145,10 @@ class MercureHandler
         $data = [];
         foreach ($body as $param) {
             if (!str_contains($param, "="))
-                throw new RequestException("Invalid request data", RequestException::ERR_INVALID_REQUEST_DATA);
+                throw new RequestException(
+                    message: "Invalid request data", 
+                    code: RequestException::ERR_INVALID_REQUEST_DATA
+                );
             [ $name, $value ] = array_map('urldecode', explode("=", $param, 2));
             if (in_array($name, [ 'topic' ])) {
                 if (!isset($data[$name]))
@@ -158,7 +167,10 @@ class MercureHandler
                 $publishClaims = $claims['mercure']['publish'];
                 // check topic against publishClaims
                 if (!$this->checkTopicClaims($data['topic']??[], $publishClaims)) {
-                    throw new SecurityException("Insufficient permissions for publish", SecurityException::ERR_NO_PERMISSION);
+                    throw new SecurityException(
+                        message: "Insufficient permissions for publish", 
+                        code: SecurityException::ERR_NO_PERMISSION
+                    );
                 }
             }
         } else {
@@ -209,24 +221,6 @@ class MercureHandler
     private function publishMercureMessage(Message $message): void
     {
         $this->topicManager->publish($message);
-
-        // foreach ($this->webSocketClients as $webSocket) {
-        //     $webSocket->write(json_encode([
-        //         'type' => $message->type,
-        //         'topic' => $message->topic,
-        //         'data' => (@json_decode($message->data))??$message->data
-        //     ]));
-        // }
-
-        $sseMessage = "";
-        if ($message->type) {
-            $sseMessage .= "event: ".$message->type."\n";
-        }
-        $sseMessage .= "data: ".$message->data."\n\n";
-
-        foreach ($this->eventClients as $client) {
-            $client->write($sseMessage);
-        }
     }
 
 }

src/Http/Server.php

@@ -42,7 +42,7 @@ class Server
      * 
      * 
      */
-    public function __construct(Configuration $config, array $options=[], ?LoopInterface $loop=null)
+    public function __construct(Configuration $config, ?LoopInterface $loop=null)
     {
         $this->loop = $loop??Loop::get();
 
@@ -53,7 +53,7 @@ class Server
         $this->eventClients = new SplObjectStorage();
         $this->webSocketClients = new SplObjectStorage();
 
-        $this->server = $this->createHttpServer($options);
+        $this->server = $this->createHttpServer();
     }
 
     /**
@@ -70,7 +70,7 @@ class Server
      * 
      * @return HttpServer
      */
-    private function createHttpServer(array $options): HttpServer
+    private function createHttpServer(): HttpServer
     {
         return new HttpServer(
             $this->responseMiddleware = new ResponseMiddleware(
@@ -85,8 +85,7 @@ class Server
                 topicManager: $this->topicManager
             ),
             $this->mercureHandler     = new MercureHandler(
-                config: $this->config, 
+                config: $this->config,
-                eventClients: $this->eventClients, 
                 topicManager: $this->topicManager
             ),
             $this->apiRequestHandler  = new ApiHandler(