noccy/mercureact
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

First unit tests, misc fixes

Browse Source
This commit is contained in:
Chris
2024-03-12 01:45:21 +01:00
parent b3476881e1
commit 8be2e81054
8 changed files with 173 additions and 22 deletions
Download Patch File Download Diff File Expand all files Collapse all files

30
src/Http/Middleware/MercureHandler.php
View File

@ -2,6 +2,7 @@
namespace NoccyLabs\Mercureact\Http\Middleware;
use LDAP\Result;
use NoccyLabs\Mercureact\Broker\Message;
use NoccyLabs\Mercureact\Broker\SseSubscriber;
use NoccyLabs\Mercureact\Broker\TopicManager;
@ -24,6 +25,10 @@ class MercureHandler
{
private LoopInterface $loop;
private array $seenMessageIds = [];
private int $seenIdHistorySize = 100;
public function __construct(
private Configuration $config,
private TopicManager $topicManager,
@ -31,6 +36,7 @@ class MercureHandler
)
{
$this->loop = $loop ?? Loop::get();
$this->seenIdHistorySize = $this->config->getDuplicateIdHistorySize();
}
/**
@ -166,9 +172,16 @@ class MercureHandler
);
}
if ($this->config->getRejectDuplicateMessages() && !empty($data['id'])) {
if (in_array($data['id'], $this->seenMessageIds)) {
return Response::plaintext("Duplicate message id")->withStatus(Response::STATUS_BAD_REQUEST);
}
array_push($this->seenMessageIds, $data['id']);
$this->seenMessageIds = array_slice($this->seenMessageIds, -100, 100);
}
// Put an id in there if none already
// TODO add a configurable for this
if (!isset($data['id'])) {
if (empty($data['id']) || $this->config->getOverwriteMessageIds()) {
$data['id'] = (string)Uuid::v7();
}
@ -176,7 +189,7 @@ class MercureHandler
$message = Message::fromData($data);
$this->loop->futureTick(function () use ($message) {
$this->publishMercureMessage($message);
$this->topicManager->publish($message);
});
return Response::plaintext("urn:uuid:".$message->id."\n");
@ -201,15 +214,4 @@ class MercureHandler
return ($matched == count($topic));
}
/**
*
*
* @param Message $message
* @return void
*/
private function publishMercureMessage(Message $message): void
{
$this->topicManager->publish($message);
}
}