Skip to content

Commit 09e148c

Browse files
makasimmakasim
committed
[dbal] fix tests.
1 parent 5316a97 commit 09e148c

8 files changed

Lines changed: 98 additions & 79 deletions

‎pkg/dbal/DbalConsumer.php‎

Lines changed: 1 addition & 23 deletions
Original file line numberDiff line numberDiff line change
@@ -5,13 +5,11 @@
55
namespace Enqueue\Dbal;
66

77
use Doctrine\DBAL\Connection;
8-
use Doctrine\DBAL\Types\Type;
98
use Interop\Queue\Consumer;
109
use Interop\Queue\Exception\InvalidMessageException;
1110
use Interop\Queue\Impl\ConsumerPollingTrait;
1211
use Interop\Queue\Message;
1312
use Interop\Queue\Queue;
14-
use Ramsey\Uuid\Uuid;
1513

1614
class DbalConsumer implements Consumer
1715
{
@@ -82,14 +80,7 @@ public function receiveNoWait(): ?Message
8280
$this->removeExpiredMessages();
8381
$this->redeliverMessages();
8482

85-
// get top message from the queue
86-
if ($message = $this->fetchMessage([$this->queue->getQueueName()], $redeliveryDelay)) {
87-
if ($message['redelivered'] || empty($message['time_to_live']) || $message['time_to_live'] > time()) {
88-
return $this->getContext()->convertMessage($message);
89-
}
90-
}
91-
92-
return null;
83+
return $this->fetchMessage([$this->queue->getQueueName()], $redeliveryDelay);
9384
}
9485

9586
/**
@@ -130,17 +121,4 @@ protected function getConnection(): Connection
130121
{
131122
return $this->dbal;
132123
}
133-
134-
private function deleteMessage(string $deliveryId): void
135-
{
136-
if (empty($deliveryId)) {
137-
throw new \LogicException(sprintf('Expected record was removed but it is not. Delivery id: "%s"', $deliveryId));
138-
}
139-
140-
$this->getConnection()->delete(
141-
$this->getContext()->getTableName(),
142-
['delivery_id' => Uuid::fromString($deliveryId)->getBytes()],
143-
['delivery_id' => Type::BINARY]
144-
);
145-
}
146124
}

‎pkg/dbal/DbalConsumerHelperTrait.php‎

Lines changed: 29 additions & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -19,8 +19,12 @@ abstract protected function getContext(): DbalContext;
1919

2020
abstract protected function getConnection(): Connection;
2121

22-
protected function fetchMessage(array $queues, int $redeliveryDelay): ?array
22+
protected function fetchMessage(array $queues, int $redeliveryDelay): ?DbalMessage
2323
{
24+
if (empty($queues)) {
25+
throw new \LogicException('Queues must not be empty.');
26+
}
27+
2428
$now = time();
2529
$deliveryId = Uuid::uuid1();
2630

@@ -34,8 +38,8 @@ protected function fetchMessage(array $queues, int $redeliveryDelay): ?array
3438
->andWhere('delivery_id IS NULL')
3539
->addOrderBy('priority', 'asc')
3640
->addOrderBy('published_at', 'asc')
41+
->setParameter('queues', $queues, Connection::PARAM_STR_ARRAY)
3742
->setParameter('delayedUntil', $now, ParameterType::INTEGER)
38-
->setParameter('queues', array_values($queues), Connection::PARAM_STR_ARRAY)
3943
->setMaxResults(1);
4044

4145
$update = $this->getConnection()->createQueryBuilder()
@@ -44,11 +48,11 @@ protected function fetchMessage(array $queues, int $redeliveryDelay): ?array
4448
->set('redeliver_after', ':redeliverAfter')
4549
->andWhere('id = :messageId')
4650
->andWhere('delivery_id IS NULL')
47-
->setParameter('deliveryId', $deliveryId->getBytes(), Type::BINARY)
51+
->setParameter('deliveryId', $deliveryId->getBytes(), Type::GUID)
4852
->setParameter('redeliverAfter', $now + $redeliveryDelay, Type::BIGINT)
4953
;
5054

51-
while (microtime(true) >= $endAt) {
55+
while (microtime(true) < $endAt) {
5256
$result = $select->execute()->fetch();
5357
if (empty($result)) {
5458
return null;
@@ -63,7 +67,7 @@ protected function fetchMessage(array $queues, int $redeliveryDelay): ?array
6367
->select('*')
6468
->from($this->getContext()->getTableName())
6569
->andWhere('delivery_id = :deliveryId')
66-
->setParameter('deliveryId', $deliveryId->getBytes(), Type::BINARY)
70+
->setParameter('deliveryId', $deliveryId->getBytes(), Type::GUID)
6771
->setMaxResults(1)
6872
->execute()
6973
->fetch()
@@ -73,7 +77,9 @@ protected function fetchMessage(array $queues, int $redeliveryDelay): ?array
7377
throw new \LogicException('There must be a message at all times at this stage but there is no a message.');
7478
}
7579

76-
return $deliveredMessage;
80+
if ($deliveredMessage['redelivered'] || empty($deliveredMessage['time_to_live']) || $deliveredMessage['time_to_live'] > time()) {
81+
return $this->getContext()->convertMessage($deliveredMessage);
82+
}
7783
}
7884
}
7985

@@ -95,7 +101,7 @@ protected function redeliverMessages(): void
95101
->andWhere('redeliver_after < :now')
96102
->andWhere('delivery_id IS NOT NULL')
97103
->setParameter(':now', time(), Type::BIGINT)
98-
->setParameter('deliveryId', null, Type::BINARY)
104+
->setParameter('deliveryId', null, Type::GUID)
99105
->setParameter('redelivered', true, Type::BOOLEAN)
100106
;
101107

@@ -112,17 +118,30 @@ protected function removeExpiredMessages(): void
112118
return;
113119
}
114120

115-
$update = $this->getConnection()->createQueryBuilder()
121+
$delete = $this->getConnection()->createQueryBuilder()
116122
->delete($this->getContext()->getTableName())
117123
->andWhere('(time_to_live IS NOT NULL) AND (time_to_live < :now)')
118124
->andWhere('delivery_id IS NULL')
119125
->andWhere('redelivered = false')
120126

121-
->setParameter(':now', (int) time(), Type::BIGINT)
127+
->setParameter(':now', time(), Type::BIGINT)
122128
;
123129

124-
$update->execute();
130+
$delete->execute();
125131

126132
$this->removeExpiredMessagesLastExecutedAt = microtime(true);
127133
}
134+
135+
private function deleteMessage(string $deliveryId): void
136+
{
137+
if (empty($deliveryId)) {
138+
throw new \LogicException(sprintf('Expected record was removed but it is not. Delivery id: "%s"', $deliveryId));
139+
}
140+
141+
$this->getConnection()->delete(
142+
$this->getContext()->getTableName(),
143+
['delivery_id' => Uuid::fromString($deliveryId)->getBytes()],
144+
['delivery_id' => Type::GUID]
145+
);
146+
}
128147
}

‎pkg/dbal/DbalContext.php‎

Lines changed: 25 additions & 20 deletions
Original file line numberDiff line numberDiff line change
@@ -142,32 +142,38 @@ public function createSubscriptionConsumer(): SubscriptionConsumer
142142
/**
143143
* @internal It must be used here and in the consumer only
144144
*/
145-
public function convertMessage(array $dbalMessage): DbalMessage
145+
public function convertMessage(array $arrayMessage): DbalMessage
146146
{
147-
/** @var DbalMessage $dbalMessageObj */
148-
$dbalMessageObj = $this->createMessage(
149-
$dbalMessage['body'],
150-
$dbalMessage['properties'] ? JSON::decode($dbalMessage['properties']) : [],
151-
$dbalMessage['headers'] ? JSON::decode($dbalMessage['headers']) : []
147+
/** @var DbalMessage $message */
148+
$message = $this->createMessage(
149+
$arrayMessage['body'],
150+
$arrayMessage['properties'] ? JSON::decode($arrayMessage['properties']) : [],
151+
$arrayMessage['headers'] ? JSON::decode($arrayMessage['headers']) : []
152152
);
153153

154-
if (isset($dbalMessage['redelivered'])) {
155-
$dbalMessageObj->setRedelivered((bool) $dbalMessage['redelivered']);
154+
if (isset($arrayMessage['id'])) {
155+
$message->setMessageId(Uuid::fromBytes($arrayMessage['id'])->toString());
156156
}
157-
if (isset($dbalMessage['priority'])) {
158-
$dbalMessageObj->setPriority((int) (-1 * $dbalMessage['priority']));
157+
if (isset($arrayMessage['queue'])) {
158+
$message->setQueue($arrayMessage['queue']);
159159
}
160-
if (isset($dbalMessage['published_at'])) {
161-
$dbalMessageObj->setPublishedAt((int) $dbalMessage['published_at']);
160+
if (isset($arrayMessage['redelivered'])) {
161+
$message->setRedelivered((bool) $arrayMessage['redelivered']);
162162
}
163-
if (isset($dbalMessage['delivery_id'])) {
164-
$dbalMessageObj->setDeliveryId(Uuid::fromBytes($dbalMessage['delivery_id'])->toString());
163+
if (isset($arrayMessage['priority'])) {
164+
$message->setPriority((int) (-1 * $arrayMessage['priority']));
165165
}
166-
if (isset($dbalMessage['redeliver_after'])) {
167-
$dbalMessageObj->setRedeliverAfter((int) $dbalMessage['redeliver_after']);
166+
if (isset($arrayMessage['published_at'])) {
167+
$message->setPublishedAt((int) $arrayMessage['published_at']);
168+
}
169+
if (isset($arrayMessage['delivery_id'])) {
170+
$message->setDeliveryId(Uuid::fromBytes($arrayMessage['delivery_id'])->toString());
171+
}
172+
if (isset($arrayMessage['redeliver_after'])) {
173+
$message->setRedeliverAfter((int) $arrayMessage['redeliver_after']);
168174
}
169175

170-
return $dbalMessageObj;
176+
return $message;
171177
}
172178

173179
/**
@@ -219,8 +225,7 @@ public function createDataBaseTable(): void
219225

220226
$table = new Table($this->getTableName());
221227

222-
$table->addColumn('id', Type::BINARY, ['length' => 16, 'fixed' => true]);
223-
$table->addColumn('human_id', Type::STRING, ['length' => 36]);
228+
$table->addColumn('id', Type::GUID, ['length' => 16, 'fixed' => true]);
224229
$table->addColumn('published_at', Type::BIGINT);
225230
$table->addColumn('body', Type::TEXT, ['notnull' => false]);
226231
$table->addColumn('headers', Type::TEXT, ['notnull' => false]);
@@ -230,7 +235,7 @@ public function createDataBaseTable(): void
230235
$table->addColumn('priority', Type::SMALLINT, ['notnull' => false]);
231236
$table->addColumn('delayed_until', Type::BIGINT, ['notnull' => false]);
232237
$table->addColumn('time_to_live', Type::BIGINT, ['notnull' => false]);
233-
$table->addColumn('delivery_id', Type::BINARY, ['length' => 16, 'fixed' => true, 'notnull' => false]);
238+
$table->addColumn('delivery_id', Type::GUID, ['length' => 16, 'fixed' => true, 'notnull' => false]);
234239
$table->addColumn('redeliver_after', Type::BIGINT, ['notnull' => false]);
235240

236241
$table->setPrimaryKey(['id']);

‎pkg/dbal/DbalMessage.php‎

Lines changed: 15 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -53,6 +53,11 @@ class DbalMessage implements Message
5353
*/
5454
private $deliveryId;
5555

56+
/**
57+
* @var string|null
58+
*/
59+
private $queue;
60+
5661
/**
5762
* Milliseconds, for example 15186054527288.
5863
*
@@ -249,4 +254,14 @@ public function setPublishedAt(int $publishedAt = null): void
249254
{
250255
$this->publishedAt = $publishedAt;
251256
}
257+
258+
public function getQueue(): ?string
259+
{
260+
return $this->queue;
261+
}
262+
263+
public function setQueue(?string $queue): void
264+
{
265+
$this->queue = $queue;
266+
}
252267
}

‎pkg/dbal/DbalProducer.php‎

Lines changed: 0 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -80,7 +80,6 @@ public function send(Destination $destination, Message $message): void
8080

8181
$dbalMessage = [
8282
'id' => $this->uuidCodec->encodeBinary($uuid),
83-
'human_id' => $uuid->toString(),
8483
'published_at' => $publishedAt,
8584
'body' => $body,
8685
'headers' => JSON::encode($message->getHeaders()),

‎pkg/dbal/DbalSubscriptionConsumer.php‎

Lines changed: 3 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -88,19 +88,17 @@ public function consume(int $timeout = 0): void
8888
$this->redeliverMessages();
8989

9090
if ($message = $this->fetchMessage($currentQueueNames, $redeliveryDelay)) {
91-
$dbalMessage = $this->getContext()->convertMessage($message);
92-
9391
/**
9492
* @var DbalConsumer
9593
* @var callable $callback
9694
*/
97-
list($consumer, $callback) = $this->subscribers[$message['queue']];
95+
list($consumer, $callback) = $this->subscribers[$message->getQueue()];
9896

99-
if (false === call_user_func($callback, $dbalMessage, $consumer)) {
97+
if (false === call_user_func($callback, $message, $consumer)) {
10098
return;
10199
}
102100

103-
unset($currentQueueNames[$message['queue']]);
101+
unset($currentQueueNames[$message->getQueue()]);
104102
} else {
105103
$currentQueueNames = [];
106104

‎pkg/dbal/Tests/DbalConsumerTest.php‎

Lines changed: 11 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -16,6 +16,7 @@
1616
use Interop\Queue\Exception\InvalidMessageException;
1717
use Interop\Queue\Message;
1818
use PHPUnit\Framework\TestCase;
19+
use Ramsey\Uuid\Uuid;
1920

2021
class DbalConsumerTest extends TestCase
2122
{
@@ -55,20 +56,22 @@ public function testAcknowledgeShouldThrowIfInstanceOfMessageIsInvalid()
5556

5657
public function testShouldDeleteMessageOnAcknowledge()
5758
{
59+
$deliveryId = Uuid::uuid1();
60+
5861
$queue = new DbalDestination('queue');
5962

6063
$message = new DbalMessage();
6164
$message->setBody('theBody');
62-
$message->setDeliveryId('foo-delivery-id');
65+
$message->setDeliveryId($deliveryId->toString());
6366

6467
$dbal = $this->createConectionMock();
6568
$dbal
6669
->expects($this->once())
6770
->method('delete')
6871
->with(
6972
'some-table-name',
70-
['delivery_id' => $message->getDeliveryId()],
71-
['delivery_id' => Type::STRING]
73+
['delivery_id' => $deliveryId->getBytes()],
74+
['delivery_id' => Type::GUID]
7275
)
7376
;
7477

@@ -124,20 +127,22 @@ public function testRejectShouldThrowIfInstanceOfMessageIsInvalid()
124127

125128
public function testShouldDeleteMessageFromQueueOnReject()
126129
{
130+
$deliveryId = Uuid::uuid1();
131+
127132
$queue = new DbalDestination('queue');
128133

129134
$message = new DbalMessage();
130135
$message->setBody('theBody');
131-
$message->setDeliveryId('foo-delivery-id');
136+
$message->setDeliveryId($deliveryId->toString());
132137

133138
$dbal = $this->createConectionMock();
134139
$dbal
135140
->expects($this->once())
136141
->method('delete')
137142
->with(
138143
'some-table-name',
139-
['delivery_id' => $message->getDeliveryId()],
140-
['delivery_id' => Type::STRING]
144+
['delivery_id' => $deliveryId->getBytes()],
145+
['delivery_id' => Type::GUID]
141146
)
142147
;
143148

0 commit comments

Comments
 (0)