vendor/gos/web-socket-bundle/src/Topic/TopicManager.php line 25

Open in your IDE?
  1. <?php declare(strict_types=1);
  3. namespace Gos\Bundle\WebSocketBundle\Topic;
  5. use Ratchet\ConnectionInterface;
  6. use Ratchet\Wamp\Topic;
  7. use Ratchet\Wamp\WampServerInterface;
  8. use Ratchet\WebSocket\WsServerInterface;
  10. /**
  11.  * @author Edu Salguero <edusalguero@gmail.com>
  12.  */
  13. class TopicManager implements WsServerInterface, WampServerInterface
  14. {
  15.     protected ?WampServerInterface $app = null;
  17.     /**
  18.      * @var array<string, Topic>
  19.      */
  20.     protected array $topicLookup = [];
  22.     /**
  23.      * @deprecated to be removed in 4.0, the dependency will be injected through the constructor instead
  24.      */
  25.     public function setWampApplication(WampServerInterface $app): void
  26.     {
  27.         trigger_deprecation('gos/web-socket-bundle', '3.7', '%s() is deprecated and will be removed in 4.0, the dependency will be injected through the constructor instead.', __METHOD__);
  29.         $this->app = $app;
  30.     }
  32.     public function onOpen(ConnectionInterface $conn): void
  33.     {
  34.         $conn->WAMP->subscriptions = new \SplObjectStorage();
  35.         $this->app->onOpen($conn);
  36.     }
  38.     public function onCall(ConnectionInterface $conn, $id, $topic, array $params): void
  39.     {
  40.         $this->app->onCall($conn, $id, $this->getTopic($topic), $params);
  41.     }
  43.     public function onSubscribe(ConnectionInterface $conn, $topic): void
  44.     {
  45.         $topicObj = $this->getTopic($topic);
  47.         if ($conn->WAMP->subscriptions->contains($topicObj)) {
  48.             return;
  49.         }
  51.         $this->topicLookup[(string) $topic]->add($conn);
  52.         $conn->WAMP->subscriptions->attach($topicObj);
  53.         $this->app->onSubscribe($conn, $topicObj);
  54.     }
  56.     /**
  57.      * @param Topic|string $topic
  58.      */
  59.     public function onUnsubscribe(ConnectionInterface $conn, $topic): void
  60.     {
  61.         $topicObj = $this->getTopic($topic);
  63.         if (!$conn->WAMP->subscriptions->contains($topicObj)) {
  64.             return;
  65.         }
  67.         $this->cleanTopic($topicObj, $conn);
  69.         $this->app->onUnsubscribe($conn, $topicObj);
  70.     }
  72.     public function onPublish(ConnectionInterface $conn, $topic, $event, array $exclude, array $eligible): void
  73.     {
  74.         $this->app->onPublish($conn, $this->getTopic($topic), $event, $exclude, $eligible);
  75.     }
  77.     public function onClose(ConnectionInterface $conn): void
  78.     {
  79.         $this->app->onClose($conn);
  81.         foreach ($this->topicLookup as $topic) {
  82.             $this->cleanTopic($topic, $conn);
  83.         }
  84.     }
  86.     public function onError(ConnectionInterface $conn, \Exception $e): void
  87.     {
  88.         $this->app->onError($conn, $e);
  89.     }
  91.     public function getSubProtocols(): array
  92.     {
  93.         if ($this->app instanceof WsServerInterface) {
  94.             return $this->app->getSubProtocols();
  95.         }
  97.         return [];
  98.     }
  100.     /**
  101.      * @param Topic|string $topic
  102.      *
  103.      * @throws \InvalidArgumentException if the $topic argument is not a supported type
  104.      */
  105.     public function getTopic($topic): Topic
  106.     {
  107.         if (!($topic instanceof Topic) && !\is_string($topic)) {
  108.             throw new \InvalidArgumentException(sprintf('The $topic argument of %s() must be an instance of %s or a string, %s was given.', __METHOD__, Topic::class, ('object' === \gettype($topic) ? 'an instance of '.\get_class($topic) : 'a '.\gettype($topic))));
  109.         }
  111.         $key = $topic instanceof Topic ? $topic->getId() : $topic;
  113.         if (!\array_key_exists($key, $this->topicLookup)) {
  114.             if ($topic instanceof Topic) {
  115.                 $this->topicLookup[$key] = $topic;
  116.             } else {
  117.                 $this->topicLookup[$key] = new Topic($topic);
  118.             }
  119.         }
  121.         return $this->topicLookup[$key];
  122.     }
  124.     protected function cleanTopic(Topic $topic, ConnectionInterface $conn): void
  125.     {
  126.         if ($conn->WAMP->subscriptions->contains($topic)) {
  127.             $conn->WAMP->subscriptions->detach($topic);
  128.         }
  130.         $this->topicLookup[$topic->getId()]->remove($conn);
  132.         if (0 === $topic->count()) {
  133.             unset($this->topicLookup[$topic->getId()]);
  134.         }
  135.     }
  136. }