Skip to content

Commit a02e3e4

Browse files
samdarkviktorprogger
samdark
and
viktorprogger
authored
Align AMQP adapter with current queue core (#127)
Co-authored-by: viktorprogger <viktorprogger@gmail.com>
1 parent 63e5744 commit a02e3e4

24 files changed

Lines changed: 603 additions & 249 deletions

‎.github/workflows/bechmark.yml‎

Lines changed: 18 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -61,10 +61,28 @@ jobs:
6161
run: docker compose build php${{ matrix.php }}
6262

6363
- name: "Baseline creation: Run PhpBench."
64+
id: baseline
6465
if: ${{ env.WITH_BENCH_BASELINE == '1' }}
66+
continue-on-error: true
6567
run: docker compose run --rm -e XDEBUG_MODE=off php${{ matrix.php }} php vendor/bin/phpbench run --report='aggregate' --tag=default
6668
working-directory: ./tests
6769

70+
- name: Disable baseline comparison when baseline creation failed
71+
if: ${{ env.WITH_BENCH_BASELINE == '1' && steps.baseline.outcome == 'failure' }}
72+
run: echo "WITH_BENCH_BASELINE=0" >> $GITHUB_ENV
73+
74+
- name: "Pull request: Checkout head."
75+
uses: actions/checkout@v4
76+
if: ${{ github.event_name == 'pull_request' }}
77+
with:
78+
ref: ${{ github.event.pull_request.head.sha }}
79+
clean: false
80+
81+
- name: "Pull request: Build"
82+
if: ${{ github.event_name == 'pull_request' }}
83+
working-directory: ./tests
84+
run: docker compose build php${{ matrix.php }}
85+
6886
- name: "Run PhpBench."
6987
working-directory: ./tests
7088
run: |

‎src/Adapter.php‎

Lines changed: 94 additions & 13 deletions
Original file line numberDiff line numberDiff line change
@@ -5,19 +5,22 @@
55
namespace Yiisoft\Queue\AMQP;
66

77
use BackedEnum;
8+
use InvalidArgumentException;
9+
use PhpAmqpLib\Exchange\AMQPExchangeType;
810
use PhpAmqpLib\Message\AMQPMessage;
911
use Throwable;
1012
use Yiisoft\Queue\Adapter\AdapterInterface;
1113
use Yiisoft\Queue\AMQP\Exception\NotImplementedException;
14+
use Yiisoft\Queue\AMQP\Settings\ExchangeSettingsInterface;
15+
use Yiisoft\Queue\AMQP\Settings\QueueSettingsInterface;
1216
use Yiisoft\Queue\Cli\LoopInterface;
13-
use Yiisoft\Queue\JobStatus;
17+
use Yiisoft\Queue\Message\DelayEnvelope;
1418
use Yiisoft\Queue\Message\MessageInterface;
1519
use Yiisoft\Queue\Message\MessageSerializerInterface;
20+
use Yiisoft\Queue\MessageStatus;
1621

1722
final class Adapter implements AdapterInterface
1823
{
19-
private ?AMQPMessage $amqpMessage = null;
20-
2124
public function __construct(
2225
private QueueProviderInterface $queueProvider,
2326
private readonly MessageSerializerInterface $serializer,
@@ -29,9 +32,8 @@ public function withChannel(BackedEnum|string $channel): self
2932
{
3033
$instance = clone $this;
3134

32-
$channelName = is_string($channel) ? $channel : (string) $channel->value;
33-
$instance->queueProvider = $this->queueProvider->withChannelName($channelName);
34-
$instance->amqpMessage = null;
35+
$queueName = is_string($channel) ? $channel : (string) $channel->value;
36+
$instance->queueProvider = $this->queueProvider->withQueueName($queueName);
3537

3638
return $instance;
3739
}
@@ -47,29 +49,30 @@ public function runExisting(callable $handlerCallback): void
4749
/**
4850
* @return never
4951
*/
50-
public function status(int|string $id): JobStatus
52+
public function status(int|string $id): MessageStatus
5153
{
5254
throw new NotImplementedException('Status check is not supported by the adapter ' . self::class . '.');
5355
}
5456

5557
public function push(MessageInterface $message): MessageInterface
5658
{
57-
$this->amqpMessage ??= new AMQPMessage(
59+
$queueProvider = $this->getQueueProviderForMessage($message);
60+
61+
$amqpMessage = new AMQPMessage(
5862
'',
59-
$this->queueProvider->getMessageProperties(),
63+
$queueProvider->getMessageProperties(),
6064
);
61-
$amqpMessage = $this->amqpMessage;
6265

6366
$payload = $this->serializer->serialize($message);
6467
$amqpMessage->setBody($payload);
65-
$exchangeSettings = $this->queueProvider->getExchangeSettings();
68+
$exchangeSettings = $queueProvider->getExchangeSettings();
6669

67-
$this->queueProvider
70+
$queueProvider
6871
->getChannel()
6972
->basic_publish(
7073
$amqpMessage,
7174
$exchangeSettings?->getName() ?? '',
72-
$exchangeSettings ? '' : $this->queueProvider
75+
$exchangeSettings ? '' : $queueProvider
7376
->getQueueSettings()
7477
->getName()
7578
);
@@ -80,6 +83,17 @@ public function push(MessageInterface $message): MessageInterface
8083
public function subscribe(callable $handlerCallback): void
8184
{
8285
$channel = $this->queueProvider->getChannel();
86+
$qosSettings = $this->queueProvider
87+
->getQueueSettings()
88+
->getQosSettings();
89+
if ($qosSettings !== null) {
90+
$channel->basic_qos(
91+
$qosSettings->getPrefetchSize(),
92+
$qosSettings->getPrefetchCount(),
93+
$qosSettings->isGlobal(),
94+
);
95+
}
96+
8397
$channel->basic_consume(
8498
$this->queueProvider
8599
->getQueueSettings()
@@ -128,4 +142,71 @@ public function getChannel(): string
128142
{
129143
return $this->queueProvider->getQueueSettings()->getName();
130144
}
145+
146+
private function getQueueProviderForMessage(MessageInterface $message): QueueProviderInterface
147+
{
148+
$delaySeconds = DelayEnvelope::fromMessage($message)->getDelaySeconds();
149+
if ($delaySeconds <= 0) {
150+
return $this->queueProvider;
151+
}
152+
153+
$exchangeSettings = $this->queueProvider->getExchangeSettings();
154+
if ($exchangeSettings === null) {
155+
throw new InvalidArgumentException('Message cannot be delayed to a queue without an exchange. Exchange is mandatory.');
156+
}
157+
158+
$delayMilliseconds = (int) ceil($delaySeconds * 1000);
159+
160+
return $this->queueProvider
161+
->withMessageProperties($this->getDelayMessageProperties($delayMilliseconds))
162+
->withExchangeSettings($this->getDelayExchangeSettings($exchangeSettings))
163+
->withQueueSettings(
164+
$this->getDelayQueueSettings(
165+
$this->queueProvider->getQueueSettings(),
166+
$exchangeSettings,
167+
$delayMilliseconds,
168+
)
169+
);
170+
}
171+
172+
/**
173+
* @psalm-return array{expiration: string, delivery_mode: int}&array
174+
*/
175+
private function getDelayMessageProperties(int $delayMilliseconds): array
176+
{
177+
return array_merge(
178+
$this->queueProvider->getMessageProperties(),
179+
[
180+
'expiration' => (string) $delayMilliseconds,
181+
'delivery_mode' => AMQPMessage::DELIVERY_MODE_PERSISTENT,
182+
],
183+
);
184+
}
185+
186+
private function getDelayQueueSettings(
187+
QueueSettingsInterface $queueSettings,
188+
ExchangeSettingsInterface $exchangeSettings,
189+
int $delayMilliseconds,
190+
): QueueSettingsInterface {
191+
$deliveryTime = time() + (int) ceil($delayMilliseconds / 1000);
192+
193+
return $queueSettings
194+
->withName("{$queueSettings->getName()}.dlx.$deliveryTime")
195+
->withAutoDeletable(true)
196+
->withArguments(
197+
[
198+
'x-dead-letter-exchange' => ['S', $exchangeSettings->getName()],
199+
'x-expires' => ['I', $delayMilliseconds + 30000],
200+
'x-message-ttl' => ['I', $delayMilliseconds],
201+
]
202+
);
203+
}
204+
205+
private function getDelayExchangeSettings(ExchangeSettingsInterface $exchangeSettings): ExchangeSettingsInterface
206+
{
207+
return $exchangeSettings
208+
->withName("{$exchangeSettings->getName()}.dlx")
209+
->withAutoDelete(true)
210+
->withType(AMQPExchangeType::TOPIC);
211+
}
131212
}

‎src/Exception/ExchangeDeclaredException.php‎

Lines changed: 4 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -17,16 +17,16 @@ public function getName(): string
1717
public function getSolution(): ?string
1818
{
1919
return <<<'SOLUTION'
20-
Can't explicitly set channel name when an exchange is declared.
20+
Can't explicitly set queue name when an exchange is declared.
2121
2222
Probably, you have called QueueFactory::get() without explicit configuration
23-
for a given channel.
23+
for a given queue.
2424
Your QueueProvider configuration has an exchange,
2525
which can't be implicitly binded to a new queue due to differences in behaviors
2626
of different types of exchanges. Please, create an explicit configuration
27-
with a fully-configured adapter for the channel you are trying to create.
27+
with a fully-configured adapter for the queue you are trying to create.
2828
29-
Reference: https://github.com/yiisoft/yii-queue#different-queue-channels
29+
Reference: https://github.com/yiisoft/queue/blob/master/docs/guide/en/queue-names.md
3030

3131
SOLUTION;
3232
}

‎src/Middleware/DelayMiddleware.php‎

Lines changed: 10 additions & 80 deletions
Original file line numberDiff line numberDiff line change
@@ -4,20 +4,14 @@
44

55
namespace Yiisoft\Queue\AMQP\Middleware;
66

7-
use InvalidArgumentException;
8-
use PhpAmqpLib\Exchange\AMQPExchangeType;
9-
use PhpAmqpLib\Message\AMQPMessage;
10-
use Yiisoft\Queue\AMQP\Adapter;
11-
use Yiisoft\Queue\AMQP\QueueProviderInterface;
12-
use Yiisoft\Queue\AMQP\Settings\ExchangeSettingsInterface;
13-
use Yiisoft\Queue\AMQP\Settings\QueueSettingsInterface;
14-
use Yiisoft\Queue\Middleware\Push\Implementation\DelayMiddlewareInterface;
15-
use Yiisoft\Queue\Middleware\Push\MessageHandlerPushInterface;
16-
use Yiisoft\Queue\Middleware\Push\PushRequest;
7+
use Yiisoft\Queue\Message\DelayEnvelope;
8+
use Yiisoft\Queue\Message\MessageInterface;
9+
use Yiisoft\Queue\Middleware\Push\PushHandlerInterface;
10+
use Yiisoft\Queue\Middleware\Push\PushMiddlewareInterface;
1711

18-
final class DelayMiddleware implements DelayMiddlewareInterface
12+
final class DelayMiddleware implements PushMiddlewareInterface
1913
{
20-
public function __construct(private float $delayInSeconds, private readonly bool $forcePersistentMessages = true)
14+
public function __construct(private float $delayInSeconds)
2115
{
2216
}
2317

@@ -39,76 +33,12 @@ public function getDelay(): float
3933
return $this->delayInSeconds;
4034
}
4135

42-
public function processPush(PushRequest $request, MessageHandlerPushInterface $handler): PushRequest
36+
public function processPush(MessageInterface $message, PushHandlerInterface $handler): MessageInterface
4337
{
44-
$adapter = $request->getAdapter();
45-
if (!$adapter instanceof Adapter) {
46-
$type = get_debug_type($adapter);
47-
$class = Adapter::class;
48-
throw new InvalidArgumentException(
49-
"This middleware works only with the $class. $type given."
50-
);
38+
if ($this->delayInSeconds <= 0) {
39+
return $handler->handlePush($message);
5140
}
5241

53-
if ($adapter->getQueueProvider()->getExchangeSettings() === null) {
54-
throw new InvalidArgumentException('Message cannot be delayed to a queue without an exchange. Exchange is mandatory.');
55-
}
56-
57-
$queueProvider = $adapter->getQueueProvider();
58-
$exchangeSettings = $this->getExchangeSettings($queueProvider->getExchangeSettings());
59-
$queueSettings = $this->getQueueSettings($queueProvider->getQueueSettings(), $queueProvider->getExchangeSettings());
60-
$adapter = $adapter->withQueueProvider(
61-
$queueProvider
62-
->withMessageProperties($this->getMessageProperties($queueProvider))
63-
->withExchangeSettings($exchangeSettings)
64-
->withQueueSettings($queueSettings)
65-
);
66-
67-
return $handler->handlePush($request->withAdapter($adapter));
68-
}
69-
70-
/**
71-
* @psalm-return array{expiration: int|float, delivery_mode?: int}&array
72-
*/
73-
private function getMessageProperties(QueueProviderInterface $queueProvider): array
74-
{
75-
$messageProperties = ['expiration' => $this->delayInSeconds * 1000];
76-
if ($this->forcePersistentMessages === true) {
77-
$messageProperties['delivery_mode'] = AMQPMessage::DELIVERY_MODE_PERSISTENT;
78-
}
79-
80-
return array_merge($queueProvider->getMessageProperties(), $messageProperties);
81-
}
82-
83-
private function getQueueSettings(
84-
QueueSettingsInterface $queueSettings,
85-
?ExchangeSettingsInterface $exchangeSettings
86-
): QueueSettingsInterface {
87-
$deliveryTime = time() + $this->delayInSeconds;
88-
89-
return $queueSettings
90-
->withName("{$queueSettings->getName()}.dlx.$deliveryTime")
91-
->withAutoDeletable(true)
92-
->withArguments(
93-
[
94-
'x-dead-letter-exchange' => ['S', $exchangeSettings?->getName() ?? ''],
95-
'x-expires' => ['I', $this->delayInSeconds * 1000 + 30000],
96-
'x-message-ttl' => ['I', $this->delayInSeconds * 1000],
97-
]
98-
);
99-
}
100-
101-
/**
102-
* @see https://github.com/vimeo/psalm/issues/9454
103-
*
104-
* @psalm-suppress LessSpecificReturnType
105-
*/
106-
private function getExchangeSettings(?ExchangeSettingsInterface $exchangeSettings): ?ExchangeSettingsInterface
107-
{
108-
/** @noinspection NullPointerExceptionInspection */
109-
return $exchangeSettings
110-
?->withName("{$exchangeSettings->getName()}.dlx")
111-
->withAutoDelete(true)
112-
->withType(AMQPExchangeType::TOPIC);
42+
return $handler->handlePush(new DelayEnvelope($message, $this->delayInSeconds));
11343
}
11444
}

‎src/QueueProvider.php‎

Lines changed: 4 additions & 16 deletions
Original file line numberDiff line numberDiff line change
@@ -50,7 +50,7 @@ public function getChannel(): AMQPChannel
5050
}
5151

5252
$this->channelId = $this->connection->get_free_channel_id();
53-
$channel = $this->connection->channel($this->getChannelId());
53+
$channel = $this->connection->channel($this->channelId);
5454
$channel->queue_declare(...$this->queueSettings->getPositionalSettings());
5555

5656
if ($this->exchangeSettings !== null) {
@@ -76,9 +76,9 @@ public function getMessageProperties(): array
7676
return $this->messageProperties;
7777
}
7878

79-
public function withChannelName(string $channel): self
79+
public function withQueueName(string $queue): self
8080
{
81-
if ($channel === $this->queueSettings->getName()) {
81+
if ($queue === $this->queueSettings->getName()) {
8282
return $this;
8383
}
8484

@@ -87,10 +87,7 @@ public function withChannelName(string $channel): self
8787
}
8888

8989
$instance = clone $this;
90-
$instance->queueSettings = $instance->queueSettings->withName($channel);
91-
if ($this->channelId !== null) {
92-
$instance->channelId = null;
93-
}
90+
$instance->queueSettings = $instance->queueSettings->withName($queue);
9491

9592
return $instance;
9693
}
@@ -135,13 +132,4 @@ public function channelClose(): void
135132
$this->channelId = null;
136133
}
137134
}
138-
139-
private function getChannelId(): int
140-
{
141-
if ($this->channelId === null) {
142-
$this->channelId = $this->connection->get_free_channel_id();
143-
}
144-
145-
return $this->channelId;
146-
}
147135
}

‎src/QueueProviderInterface.php‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -21,7 +21,7 @@ public function getExchangeSettings(): ?ExchangeSettingsInterface;
2121

2222
public function getMessageProperties(): array;
2323

24-
public function withChannelName(string $channel): self;
24+
public function withQueueName(string $queue): self;
2525

2626
public function withQueueSettings(QueueSettingsInterface $queueSettings): self;
2727

0 commit comments

Comments
 (0)