Fix: Unsubscribe from transport on closing connection (#12)

src/Hub/Controller/SubscribeController.php

@@ -69,9 +69,14 @@ function () use ($lastEventId, $stream, $subscribedTopics, $allowedTopics) {
 
  async(
  function () use ($stream, $subscribedTopics, $allowedTopics) {
- $this->hub->subscribe(function (Update $update) use ($stream, $subscribedTopics, $allowedTopics) {
- $this->sendUpdate($update, $stream, $subscribedTopics, $allowedTopics);
- });
+ $callback = fn(Update $update) => $this->sendUpdate(
+ $update,
+ $stream,
+ $subscribedTopics,
+ $allowedTopics
+ );
+ $this->hub->subscribe($callback);
+ $stream->on('close', fn() => $this->hub->unsubscribe($callback));
  }
  )();
 

src/Hub/Hub.php

@@ -80,6 +80,11 @@ public function subscribe(callable $callback): void
  $this->transport->subscribe($callback);
  }
 
+ public function unsubscribe(callable $callback): void
+ {
+ $this->transport->unsubscribe($callback);
+ }
+
  public function reconciliate(string $lastEventID): Generator
  {
  return $this->transport->reconciliate($lastEventID);

src/Hub/Transport/PHP/PHPTransport.php

@@ -41,6 +41,11 @@ public function subscribe(callable $callback): void
  $this->eventEmitter->on('mercureUpdate', $callback);
  }
 
+ public function unsubscribe(callable $callback): void
+ {
+ $this->eventEmitter->removeListener('mercureUpdate', $callback);
+ }
+
  public function reconciliate(string $lastEventID): Generator
  {
  $yield = self::EARLIEST === $lastEventID;

src/Hub/Transport/Redis/RedisTransport.php

@@ -5,6 +5,8 @@
 namespace Freddie\Hub\Transport\Redis;
 
 use Clue\React\Redis\Client;
+use Evenement\EventEmitter;
+use Evenement\EventEmitterInterface;
 use Freddie\Hub\Transport\TransportInterface;
 use Freddie\Message\Update;
 use Generator;
@@ -24,6 +26,7 @@ public function __construct(
  public readonly Client $subscriber,
  public readonly Client $redis,
  private RedisSerializer $serializer = new RedisSerializer(),
+ private EventEmitterInterface $eventEmitter = new EventEmitter(),
  private int $size = 0,
  private float $trimInterval = 0.0,
  ) {
@@ -32,9 +35,12 @@ public function __construct(
  public function subscribe(callable $callback): void
  {
  $this->init();
- $this->subscriber->on('message', function (string $channel, string $payload) use ($callback) {
- $callback($this->serializer->deserialize($payload));
- });
+ $this->eventEmitter->on('mercureUpdate', $callback);
+ }
+
+ public function unsubscribe(callable $callback): void
+ {
+ $this->eventEmitter->removeListener('mercureUpdate', $callback);
  }
 
  public function publish(Update $update): PromiseInterface
@@ -84,6 +90,10 @@ private function init(): void
  }
 
  $this->subscriber->subscribe($this->channel); // @phpstan-ignore-line
+ $this->subscriber->on('message', function (string $channel, string $payload) {
+ $this->eventEmitter->emit('mercureUpdate', [$this->serializer->deserialize($payload)]);
+ });
+
  if ($this->trimInterval > 0) {
  Loop::addPeriodicTimer(
  $this->trimInterval,

src/Hub/Transport/TransportInterface.php

@@ -19,6 +19,8 @@ public function publish(Update $update): PromiseInterface;
 
  public function subscribe(callable $callback): void;
 
+ public function unsubscribe(callable $callback): void;
+
  /**
  * @param string $lastEventID
  * @return Generator<Update>

tests/Unit/Hub/Controller/SubscribeControllerTest.php

@@ -161,3 +161,36 @@
  AccessDeniedHttpException::class,
  'Error while decoding from Base64Url, invalid base64 characters detected'
 );
+
+
+it('unsubscribes from transport whenever connection closes', function () {
+ $transport = new PHPTransport(size: 1000);
+ $controller = new SubscribeController();
+ $controller->setHub(new Hub(transport: $transport));
+ $stream = new ThroughStreamStub();
+
+ // Given
+ $hey = new Message(data: 'Hey!');
+ $hello = new Message(data: 'Hello');
+ $world = new Message(data: 'World!');
+ $transport->publish(new Update(['/bar'], $hey)); // Should not be dumped into stream
+ $transport->publish(new Update(['/foo'], $hello));
+ $request = new ServerRequest(
+ 'GET',
+ '/.well-known/mercure?topic=/foo',
+ ['Last-Event-ID' => 'earliest'],
+ );
+
+ // When
+ $response = $controller($request, $stream);
+ Loop::addTimer(0.01, fn () => Loop::stop());
+ Loop::futureTick(fn () => $stream->close());
+ Loop::futureTick(fn () => $transport->publish(new Update(['/foo'], $world)));
+ Loop::run();
+
+ // Then
+ expect($response->getStatusCode())->toBe(200);
+ expect($response->getHeaderLine('Content-Type'))->toBe('text/event-stream');
+ expect($stream->storage)->toHaveCount(1);
+ expect($stream->storage[0])->toBe((string) $hello);
+});

tests/Unit/Hub/Controller/ThroughStreamStub.php

@@ -53,6 +53,6 @@ public function end($data = null): void
 
  public function close(): void
  {
- throw new ShouldNotHappen(new LogicException(__METHOD__));
+ $this->emit('close');
  }
 }

tests/Unit/Hub/HubTest.php

@@ -32,6 +32,11 @@ public function subscribe(callable $callback): void
  $this->called['subscribe'] = func_get_args();
  }
 
+ public function unsubscribe(callable $callback): void
+ {
+ $this->called['unsubscribe'] = func_get_args();
+ }
+
  public function reconciliate(string $lastEventID): Generator
  {
  $this->called['reconciliate'] = func_get_args();
@@ -49,11 +54,13 @@ public function reconciliate(string $lastEventID): Generator
  $hub->publish($update);
  $hub->subscribe($subscribeFn);
  iterator_to_array($hub->reconciliate($lastEventId));
+ $hub->unsubscribe($subscribeFn);
 
  // Then
  expect($transport->called['publish'])->toBe([$update]);
  expect($transport->called['subscribe'])->toBe([$subscribeFn]);
  expect($transport->called['reconciliate'])->toBe([$lastEventId]);
+ expect($transport->called['unsubscribe'])->toBe([$subscribeFn]);
 });
 
 it('complains when requesting an unrecognized option', function () {

tests/Unit/Hub/Transport/PHP/PHPTransportTest.php

@@ -15,8 +15,20 @@
  $transport = new PHPTransport();
  $update = new Update(['/foo'], new Message());
  $subscriber = (object) ['received' => null];
- $transport->subscribe(fn ($receivedUpdate) => $subscriber->received = $receivedUpdate);
+ $callback = fn($receivedUpdate) => $subscriber->received = $receivedUpdate;
+ $transport->subscribe($callback);
+
+ // When
  $transport->publish($update);
+
+ // Then
+ expect($subscriber->received ?? null)->toBe($update);
+
+ // When
+ $transport->unsubscribe($callback);
+ $transport->publish(new Update(['/foo'], new Message('bar')));
+
+ // Then
  expect($subscriber->received ?? null)->toBe($update);
 });
 

tests/Unit/Hub/Transport/Redis/RedisTransportTest.php

@@ -22,14 +22,20 @@
  // Given
  $subscriber = (object) ['received' => null];
  $update = new Update(['/foo'], new Message('bar'));
+ $callback = fn($receivedUpdate) => $subscriber->received = $receivedUpdate;
 
  // When
- $transport->subscribe(fn($receivedUpdate) => $subscriber->received = $receivedUpdate);
+ $transport->subscribe($callback);
  $transport->publish($update);
 
  // Then
  expect($subscriber->received ?? null)->not()->toBe($update); // Because serialization/deserialization
  expect($subscriber->received ?? null)->toEqual($update);
+
+ // When
+ $transport->unsubscribe($callback);
+ $transport->publish(new Update(['/foo'], new Message('foobar')));
+ expect($subscriber->received ?? null)->toEqual($update);
 });
 
 it('performs state reconciliation', function () {