refactor: introduce subscribers

src/Hub/Controller/SubscribeController.php

@@ -8,6 +8,7 @@
 use Freddie\Hub\HubControllerInterface;
 use Freddie\Hub\HubInterface;
 use Freddie\Message\Update;
+use Freddie\Subscription\Subscriber;
 use Lcobucci\JWT\UnencryptedToken;
 use Psr\Http\Message\ResponseInterface;
 use Psr\Http\Message\ServerRequestInterface;
@@ -57,6 +58,8 @@ public function __invoke(
  $allowedTopics = $this->extractAllowedTopics($request);
  $lastEventId = extract_last_event_id($request);
 
+ $subscriber = new Subscriber($subscribedTopics);
+
  if (null !== $lastEventId) {
  async(
  function () use ($lastEventId, $stream, $subscribedTopics, $allowedTopics) {
@@ -68,15 +71,16 @@ function () use ($lastEventId, $stream, $subscribedTopics, $allowedTopics) {
  }
 
  async(
- function () use ($stream, $subscribedTopics, $allowedTopics) {
+ function () use ($stream, $subscribedTopics, $allowedTopics, $subscriber) {
  $callback = fn(Update $update) => $this->sendUpdate(
  $update,
  $stream,
  $subscribedTopics,
  $allowedTopics
  );
- $this->hub->subscribe($callback);
- $stream->on('close', fn() => $this->hub->unsubscribe($callback));
+ $subscriber->setCallback($callback);
+ $this->hub->subscribe($subscriber);
+ $stream->on('close', fn() => $this->hub->unsubscribe($subscriber));
  }
  )();
 

src/Hub/Hub.php

@@ -9,6 +9,7 @@
 use Freddie\Hub\Transport\PHP\PHPTransport;
 use Freddie\Hub\Transport\TransportInterface;
 use Freddie\Message\Update;
+use Freddie\Subscription\Subscriber;
 use Generator;
 use InvalidArgumentException;
 use React\EventLoop\Loop;
@@ -75,14 +76,14 @@ public function publish(Update $update): PromiseInterface
  });
  }
 
- public function subscribe(callable $callback): void
+ public function subscribe(Subscriber $subscriber): void
  {
- $this->transport->subscribe($callback);
+ $this->transport->subscribe($subscriber);
  }
 
- public function unsubscribe(callable $callback): void
+ public function unsubscribe(Subscriber $subscriber): void
  {
- $this->transport->unsubscribe($callback);
+ $this->transport->unsubscribe($subscriber);
  }
 
  public function reconciliate(string $lastEventID): Generator

src/Hub/HubInterface.php

@@ -5,6 +5,7 @@
 namespace Freddie\Hub;
 
 use Freddie\Message\Update;
+use Freddie\Subscription\Subscriber;
 use Generator;
 use React\Promise\PromiseInterface;
 
@@ -17,9 +18,9 @@ public function getOption(string $name): mixed;
  */
  public function publish(Update $update): PromiseInterface;
 
- public function subscribe(callable $callback): void;
+ public function subscribe(Subscriber $subscriber): void;
 
- public function unsubscribe(callable $callback): void;
+ public function unsubscribe(Subscriber $subscriber): void;
 
  /**
  * @param string $lastEventID

src/Message/Message.php

@@ -4,6 +4,8 @@
 
 namespace Freddie\Message;
 
+use Symfony\Component\Uid\Ulid;
+
 use function explode;
 use function str_contains;
 
@@ -18,6 +20,7 @@ public function __construct(
  public ?string $event = null,
  public ?int $retry = null,
  ) {
+ $this->id ??= (string) new Ulid();
  }
 
  public function __toString(): string

src/Message/Update.php

@@ -7,16 +7,23 @@
 use Freddie\Helper\TopicHelper;
 
 use function Freddie\topic;
+use function is_string;
 
 final class Update
 {
+ /**
+ * @var string[]
+ */
+ public array $topics;
+
  /**
  * @param string[] $topics
  */
  public function __construct(
- public array $topics,
+ array|string $topics,
  public Message $message,
  ) {
+ $this->topics = is_string($topics) ? [$topics] : $topics;
  }
 
  /**

src/Subscription/Subscriber.php

@@ -0,0 +1,50 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Freddie\Subscription;
+
+use Symfony\Component\Uid\Ulid;
+
+use function array_map;
+
+final class Subscriber
+{
+ /**
+ * @var Subscription[]
+ */
+ public readonly array $subscriptions;
+
+ /**
+ * @var callable
+ */
+ private $callback;
+
+ /**
+ * @param string[] $topics
+ */
+ public function __construct(
+ public readonly array $topics,
+ public readonly mixed $payload = null,
+ public readonly Ulid $id = new Ulid(),
+ public bool $active = true,
+ ) {
+ $this->subscriptions = array_map(
+ fn(string $topic) => new Subscription($this, $topic),
+ $this->topics,
+ );
+ }
+
+ public function setCallback(callable $callback): void
+ {
+ $this->callback = $callback;
+ }
+
+ /**
+ * @param mixed ...$args
+ */
+ public function __invoke(...$args): void
+ {
+ ($this->callback)(...$args);
+ }
+}

src/Subscription/Subscription.php

@@ -0,0 +1,14 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Freddie\Subscription;
+
+final class Subscription
+{
+ public function __construct(
+ public readonly Subscriber $subscriber,
+ public readonly string $topic,
+ ) {
+ }
+}

tests/Unit/Hub/HubTest.php

@@ -8,6 +8,7 @@
 use Freddie\Hub\Transport\TransportInterface;
 use Freddie\Message\Message;
 use Freddie\Message\Update;
+use Freddie\Subscription\Subscriber;
 use Generator;
 use InvalidArgumentException;
 use React\Promise\PromiseInterface;
@@ -48,19 +49,21 @@ public function reconciliate(string $lastEventID): Generator
  $hub = new Hub(transport: $transport);
  $update = new Update(['foo'], new Message(Ulid::generate()));
  $subscribeFn = fn () => 'bar';
+ $subscriber = new Subscriber(['foo']);
+ $subscriber->setCallback($subscribeFn);
  $lastEventId = Ulid::generate();
 
  // When
  $hub->publish($update);
- $hub->subscribe($subscribeFn);
+ $hub->subscribe($subscriber);
  iterator_to_array($hub->reconciliate($lastEventId));
- $hub->unsubscribe($subscribeFn);
+ $hub->unsubscribe($subscriber);
 
  // Then
  expect($transport->called['publish'])->toBe([$update]);
- expect($transport->called['subscribe'])->toBe([$subscribeFn]);
+ expect($transport->called['subscribe'])->toBe([$subscriber]);
  expect($transport->called['reconciliate'])->toBe([$lastEventId]);
- expect($transport->called['unsubscribe'])->toBe([$subscribeFn]);
+ expect($transport->called['unsubscribe'])->toBe([$subscriber]);
 });
 
 it('complains when requesting an unrecognized option', function () {

tests/Unit/Message/MessageTest.php

@@ -5,6 +5,7 @@
 namespace Freddie\Tests\Unit\Message;
 
 use Freddie\Message\Message;
+use Symfony\Component\Uid\Ulid;
 
 it('stringifies messages', function (Message $message, string $expected) {
  expect((string) $message)->toBe($expected);
@@ -34,3 +35,9 @@
  "id:1\nevent:message\ndata:foo\ndata:bar\n\n",
  ];
 });
+
+it('has a default id', function () {
+ $message = new Message();
+ expect($message->id)->not()->toBeNull();
+ expect(Ulid::isValid($message->id))->toBeTrue();
+});

tests/Unit/Message/UpdateTest.php

@@ -103,3 +103,8 @@
  'expected' => true, // Private update allowed on all
  ];
 });
+
+it('accepts a single topic', function () {
+ $update = new Update('/foo', new Message());
+ expect($update->topics)->toBe(['/foo']);
+});