From 9b5e7f32bae5fdd4432ee9fa3efb9d9514949fed Mon Sep 17 00:00:00 2001 From: kafkiansky Date: Thu, 30 Apr 2026 10:43:35 +0300 Subject: [PATCH 1/2] Implement AMQP like topics --- README.md | 101 +++++++++++++++++++++++++++++++++++++ docker-compose.yaml | 2 +- examples/topic.php | 36 ++++++++++++++ src/TopicRoute.php | 41 +++++++++++++++ src/pgmq.php | 108 ++++++++++++++++++++++++++++++++++++++++ tests/PgmqTest.php | 119 +++++++++++++++++++++++++++++++++++++++++--- 6 files changed, 399 insertions(+), 8 deletions(-) create mode 100644 examples/topic.php create mode 100644 src/TopicRoute.php diff --git a/README.md b/README.md index 9da0fc9..e3f3b9f 100644 --- a/README.md +++ b/README.md @@ -41,6 +41,13 @@ And only the consumer accepts `Amp\Postgres\PostgresConnection`, because it itse - [Delete batch](#delete-batch) - [Enable notify insert](#enable-notify-insert) - [Disable notify insert](#disable-notify-insert) + - [Bind topic](#bind-topic) + - [Unbind topic](#unbind-topic) + - [Send topic](#send-topic) + - [Send topic with delay](#send-topic-with-delay) + - [Test routing](#test-routing) + - [Validate routing key](#validate-routing-key) + - [Validate topic pattern](#validate-topic-pattern) - [Consume messages](#consume-messages) ### Create queue @@ -408,6 +415,100 @@ $queue = Pgmq\createQueue($pg, 'events'); $queue->disableNotifyInsert(); ``` +### Bind topic + +Bind a queue to a topic pattern. Messages sent with a routing key matching the pattern will be delivered to the queue. + +```php +use Thesis\Pgmq; +use Amp\Postgres; + +$pg = new Postgres\PostgresConnectionPool(Postgres\PostgresConfig::fromString('')); + +$queue = Pgmq\createQueue($pg, 'emails'); +Pgmq\bindTopic($pg, 'notifications.*', $queue->name); +``` + +### Unbind topic + +Remove a queue binding from a topic pattern. + +```php +use Thesis\Pgmq; +use Amp\Postgres; + +$pg = new Postgres\PostgresConnectionPool(Postgres\PostgresConfig::fromString('')); + +Pgmq\unbindTopic($pg, 'notifications.*', 'emails'); +``` + +### Send topic + +Send a message to all queues bound to patterns matching the given routing key. + +```php +use Thesis\Pgmq; +use Amp\Postgres; + +$pg = new Postgres\PostgresConnectionPool(Postgres\PostgresConfig::fromString('')); + +$matched = Pgmq\sendTopic($pg, 'notifications.email', new Pgmq\SendMessage('{"user": 1}')); +``` + +### Send topic with delay + +```php +use Thesis\Pgmq; +use Amp\Postgres; +use Thesis\Time\TimeSpan; + +$pg = new Postgres\PostgresConnectionPool(Postgres\PostgresConfig::fromString('')); + +$matched = Pgmq\sendTopic( + $pg, + 'notifications.email', + new Pgmq\SendMessage('{"user": 1}'), + TimeSpan::fromSeconds(5), +); +``` + +### Test routing + +Test which queues would receive a message for a given routing key without actually sending a message. + +```php +use Thesis\Pgmq; +use Amp\Postgres; + +$pg = new Postgres\PostgresConnectionPool(Postgres\PostgresConfig::fromString('')); + +foreach (Pgmq\testRouting($pg, 'notifications.email') as $route) { + var_dump($route->pattern, $route->queue, $route->compiledRegex); +} +``` + +### Validate routing key + +```php +use Thesis\Pgmq; +use Amp\Postgres; + +$pg = new Postgres\PostgresConnectionPool(Postgres\PostgresConfig::fromString('')); + +Pgmq\validateRoutingKey($pg, 'events.created'); // throws PostgresQueryError if invalid +``` + +### Validate topic pattern + +```php +use Thesis\Pgmq; +use Amp\Postgres; + +$pg = new Postgres\PostgresConnectionPool(Postgres\PostgresConfig::fromString('')); + +Pgmq\validateTopicPattern($pg, 'events.*'); // throws PostgresQueryError if invalid +``` + ### Consume messages This functionality is not a standard feature of the **pgmq** extension, but is provided by the library as an add-on for reliable and correct processing of message batches from the queue, with the ability to `ack`, `nack` (with delay) and archive (`term`) messages from the queue. diff --git a/docker-compose.yaml b/docker-compose.yaml index 5f8b98c..dd4cef6 100644 --- a/docker-compose.yaml +++ b/docker-compose.yaml @@ -7,7 +7,7 @@ services: command: sh -c 'trap "exit 0" TERM; tail -f /dev/null & wait' pgmq: - image: ghcr.io/pgmq/pg18-pgmq:v1.8.0 + image: ghcr.io/pgmq/pg18-pgmq:v1.11.1 environment: - POSTGRES_PASSWORD=postgres ports: diff --git a/examples/topic.php b/examples/topic.php new file mode 100644 index 0000000..88e6614 --- /dev/null +++ b/examples/topic.php @@ -0,0 +1,36 @@ +name); +Pgmq\bindTopic($pg, 'notifications.urgent', $smsQueue->name); + +// Route to all queues matching the pattern +Pgmq\sendTopic($pg, 'notifications.urgent', new Pgmq\SendMessage('{"user": 1, "text": "Server is down"}')); + +dump($emailQueue->pop()); +dump($smsQueue->pop()); + +// Route only to email queue +Pgmq\sendTopic($pg, 'notifications.welcome', new Pgmq\SendMessage('{"user": 2, "text": "Welcome!"}')); + +dump($emailQueue->pop()); +dump($smsQueue->pop()); + +Pgmq\unbindTopic($pg, 'notifications.*', $emailQueue->name); +Pgmq\unbindTopic($pg, 'notifications.urgent', $smsQueue->name); + +$emailQueue->drop(); +$smsQueue->drop(); diff --git a/src/TopicRoute.php b/src/TopicRoute.php new file mode 100644 index 0000000..12e1640 --- /dev/null +++ b/src/TopicRoute.php @@ -0,0 +1,41 @@ + $row + */ + public static function fromArray(array $row): self + { + /** @var RawRoute $row */ + return new self( + pattern: $row['pattern'], + queue: $row['queue_name'], + compiledRegex: $row['compiled_regex'], + ); + } + + /** + * @param non-empty-string $pattern + * @param non-empty-string $queue + * @param non-empty-string $compiledRegex + */ + public function __construct( + public string $pattern, + public string $queue, + public string $compiledRegex, + ) {} +} diff --git a/src/pgmq.php b/src/pgmq.php index dded62b..674caf0 100644 --- a/src/pgmq.php +++ b/src/pgmq.php @@ -194,6 +194,114 @@ function queueMetadata( return QueueMetadata::fromArray($result); } +/** + * @api + * @param non-empty-string $pattern + * @param non-empty-string $queue + */ +function bindTopic( + PostgresLink $pg, + string $pattern, + string $queue, +): void { + $pg->execute('SELECT pgmq.bind_topic(:pattern, :queue_name);', [ + 'pattern' => $pattern, + 'queue_name' => $queue, + ]); +} + +/** + * @api + * @param non-empty-string $pattern + * @param non-empty-string $queue + */ +function unbindTopic( + PostgresLink $pg, + string $pattern, + string $queue, +): void { + $pg->execute('SELECT pgmq.unbind_topic(:pattern, :queue_name);', [ + 'pattern' => $pattern, + 'queue_name' => $queue, + ]); +} + +/** + * @api + * @param non-empty-string $routingKey + * @return non-negative-int + */ +function sendTopic( + PostgresLink $pg, + string $routingKey, + SendMessage $message, + null|TimeSpan|\DateTimeImmutable $delay = null, +): int { + $delay ??= TimeSpan::fromSeconds(0); + + if ($delay instanceof \DateTimeImmutable) { + $delay = TimeSpan::diff($delay, new \DateTimeImmutable()); + } + + /** @var array{send_topic: non-negative-int} $result */ + $result = $pg + ->execute('SELECT * FROM pgmq.send_topic(:routing_key, :msg, :headers, :delay::int)', [ + 'routing_key' => $routingKey, + 'msg' => $message->valueJson, + 'headers' => $message->headerJson, + 'delay' => max((int) $delay->toSeconds(), 0), + ]) + ->fetchRow() ?? throw new \RuntimeException("Failed to send message using routingKey {$routingKey}."); + + return $result['send_topic']; +} + +/** + * @api + * @param non-empty-string $routingKey + * @return iterable + */ +function testRouting( + PostgresLink $pg, + string $routingKey, +): iterable { + $result = $pg->execute('SELECT * FROM pgmq.test_routing(:routing_key)', [ + 'routing_key' => $routingKey, + ]); + + foreach ($result as $row) { + yield TopicRoute::fromArray($row); + } +} + +/** + * @api + * @param non-empty-string $routingKey + * @throws PostgresQueryError if routingKey is invalid + */ +function validateRoutingKey( + PostgresLink $pg, + string $routingKey, +): void { + $pg->execute('SELECT * FROM pgmq.validate_routing_key(:routing_key)', [ + 'routing_key' => $routingKey, + ]); +} + +/** + * @api + * @param non-empty-string $pattern + * @throws PostgresQueryError if topic pattern is invalid + */ +function validateTopicPattern( + PostgresLink $pg, + string $pattern, +): void { + $pg->execute('SELECT * FROM pgmq.validate_topic_pattern(:pattern)', [ + 'pattern' => $pattern, + ]); +} + /** * @api * @return iterable diff --git a/tests/PgmqTest.php b/tests/PgmqTest.php index c6973dc..a88ec63 100644 --- a/tests/PgmqTest.php +++ b/tests/PgmqTest.php @@ -44,8 +44,8 @@ public function testValidateQueueName(): void { validateQueueName($this->pg, $this->randomQueueName()); - self::expectException(PostgresQueryError::class); - self::expectExceptionMessage('queue name is too long, maximum length is 47 characters'); + $this->expectException(PostgresQueryError::class); + $this->expectExceptionMessage('queue name is too long, maximum length is 47 characters'); validateQueueName($this->pg, $this->randomQueueName() . $this->randomQueueName()); } @@ -70,7 +70,7 @@ public function testCreateQueue(): void $queue->drop(); - self::expectException(QueueNotFound::class); + $this->expectException(QueueNotFound::class); $queue->metadata(); } @@ -280,8 +280,8 @@ public function testInvalidConsumerConfiguration(): void $queue = createQueue($this->pg, $this->randomQueueName()); $consumer = createConsumer($this->pg); - self::expectException(\LogicException::class); - self::expectExceptionMessage('Pooling is required. Set $pollInterval to a positive value.'); + $this->expectException(\LogicException::class); + $this->expectExceptionMessage('Pooling is required. Set $pollInterval to a positive value.'); $consumer->consume(static fn() => null, new ConsumeConfig( queue: $queue->name, pollInterval: TimeSpan::fromSeconds(0), @@ -387,8 +387,8 @@ public function testStopConsumeOnUnhandledException(): void send($this->pg, $queue->name, new SendMessage(self::TESTING_MESSAGE)); send($this->pg, $queue->name, new SendMessage(self::TESTING_MESSAGE)); - self::expectException(\RuntimeException::class); - self::expectExceptionMessage('Exception from consumer'); + $this->expectException(\RuntimeException::class); + $this->expectExceptionMessage('Exception from consumer'); $context->awaitCompletion(); } @@ -414,6 +414,111 @@ static function (array $messages, ConsumeController $c): void {}, $ctx->awaitCompletion(); } + public function testBindAndSendTopic(): void + { + $queue = createQueue($this->pg, $this->randomQueueName()); + + bindTopic($this->pg, 'events.*', $queue->name); + + $messageId = sendTopic($this->pg, 'events.created', new SendMessage(self::TESTING_MESSAGE, self::TESTING_HEADERS)); + + $message = $queue->read(); + self::assertNotNull($message); + self::assertSame($messageId, $message->id); + self::assertSame(self::TESTING_MESSAGE, $message->value); + self::assertSame(self::TESTING_HEADERS, $message->headers); + } + + public function testSendTopicWithDelay(): void + { + $queue = createQueue($this->pg, $this->randomQueueName()); + + bindTopic($this->pg, 'orders.*', $queue->name); + + sendTopic($this->pg, 'orders.placed', new SendMessage(self::TESTING_MESSAGE), TimeSpan::fromSeconds(1)); + + self::assertNull($queue->read()); + + delay(1.05); + + $message = $queue->read(); + self::assertNotNull($message); // @phpstan-ignore staticMethod.impossibleType + self::assertSame(self::TESTING_MESSAGE, $message->value); + } + + public function testSendTopicToMultipleQueues(): void + { + $queue1 = createQueue($this->pg, $this->randomQueueName()); + $queue2 = createQueue($this->pg, $this->randomQueueName()); + + bindTopic($this->pg, 'notifications.*', $queue1->name); + bindTopic($this->pg, 'notifications.*', $queue2->name); + + $messages = sendTopic($this->pg, 'notifications.email', new SendMessage(self::TESTING_MESSAGE)); + + self::assertSame(2, $messages); + $message1 = $queue1->read(); + self::assertNotNull($message1); + self::assertSame(self::TESTING_MESSAGE, $message1->value); + + $message2 = $queue2->read(); + self::assertNotNull($message2); + self::assertSame(self::TESTING_MESSAGE, $message2->value); + } + + public function testUnbindTopic(): void + { + $queue = createQueue($this->pg, $this->randomQueueName()); + + bindTopic($this->pg, 'events.*', $queue->name); + + unbindTopic($this->pg, 'events.*', $queue->name); + + sendTopic($this->pg, 'events.created', new SendMessage(self::TESTING_MESSAGE)); + + self::assertNull($queue->read()); + } + + public function testTestRouting(): void + { + $queue1 = createQueue($this->pg, $this->randomQueueName()); + $queue2 = createQueue($this->pg, $this->randomQueueName()); + + bindTopic($this->pg, 'events.*', $queue1->name); + bindTopic($this->pg, 'events.created', $queue2->name); + + $routes = [...testRouting($this->pg, 'events.created')]; + + self::assertCount(2, $routes); + + $queueNames = array_map(static fn(TopicRoute $route): string => $route->queue, $routes); + self::assertContains($queue1->name, $queueNames); + self::assertContains($queue2->name, $queueNames); + + foreach ($routes as $route) { + self::assertNotEmpty($route->pattern); + self::assertNotEmpty($route->compiledRegex); + } + } + + public function testValidateRoutingKey(): void + { + validateRoutingKey($this->pg, 'events.created'); + + $this->expectException(PostgresQueryError::class); + + validateRoutingKey($this->pg, 'events.*'); + } + + public function testValidateTopicPattern(): void + { + validateTopicPattern($this->pg, 'events.*'); + + $this->expectException(PostgresQueryError::class); + + validateTopicPattern($this->pg, 'logs.**'); + } + /** * @return non-empty-string */ From 7f1fd3fe2bf3f29a812cab327cfff90c4eaebd71 Mon Sep 17 00:00:00 2001 From: kafkiansky Date: Thu, 30 Apr 2026 10:47:24 +0300 Subject: [PATCH 2/2] chore: update ghcr.io/pgmq/pg18-pgmq in github workflows --- .github/workflows/test.yml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/.github/workflows/test.yml b/.github/workflows/test.yml index 691373c..eb7f904 100644 --- a/.github/workflows/test.yml +++ b/.github/workflows/test.yml @@ -19,7 +19,7 @@ jobs: --name pgmq-postgres \ -p 5432:5432 \ -e POSTGRES_PASSWORD=postgres \ - ghcr.io/pgmq/pg18-pgmq:v1.8.0 + ghcr.io/pgmq/pg18-pgmq:v1.11.1 - uses: shivammathur/setup-php@v2 with: php-version: ${{ matrix.php }}