|
11 | 11 |
|
12 | 12 | trait DbalConsumerHelperTrait |
13 | 13 | { |
| 14 | + private $redeliverMessagesLastExecutedAt; |
| 15 | + |
| 16 | + private $removeExpiredMessagesLastExecutedAt; |
| 17 | + |
14 | 18 | abstract protected function getContext(): DbalContext; |
15 | 19 |
|
16 | 20 | abstract protected function getConnection(): Connection; |
17 | 21 |
|
18 | | - protected function fetchMessage(array $queues, int $redeliveryDelay): ?array |
| 22 | + protected function fetchMessage(array $queues, int $redeliveryDelay): ?DbalMessage |
19 | 23 | { |
| 24 | + if (empty($queues)) { |
| 25 | + throw new \LogicException('Queues must not be empty.'); |
| 26 | + } |
| 27 | + |
20 | 28 | $now = time(); |
21 | | - $deliveryId = (string) Uuid::uuid1(); |
22 | | - |
23 | | - $this->getConnection()->beginTransaction(); |
24 | | - |
25 | | - try { |
26 | | - $query = $this->getConnection()->createQueryBuilder() |
27 | | - ->select('*') |
28 | | - ->from($this->getContext()->getTableName()) |
29 | | - ->andWhere('delivery_id IS NULL') |
30 | | - ->andWhere('delayed_until IS NULL OR delayed_until <= :delayedUntil') |
31 | | - ->andWhere('queue IN (:queues)') |
32 | | - ->addOrderBy('priority', 'desc') |
33 | | - ->addOrderBy('published_at', 'asc') |
34 | | - ->setMaxResults(1); |
35 | | - |
36 | | - // select for update |
37 | | - $message = $this->getConnection()->executeQuery( |
38 | | - $query->getSQL().' '.$this->getConnection()->getDatabasePlatform()->getWriteLockSQL(), |
39 | | - ['delayedUntil' => $now, 'queues' => array_values($queues)], |
40 | | - ['delayedUntil' => ParameterType::INTEGER, 'queues' => Connection::PARAM_STR_ARRAY] |
41 | | - )->fetch(); |
42 | | - |
43 | | - if (!$message) { |
44 | | - $this->getConnection()->commit(); |
| 29 | + $deliveryId = Uuid::uuid4(); |
| 30 | + |
| 31 | + $endAt = microtime(true) + 0.2; // add 200ms |
| 32 | + |
| 33 | + $select = $this->getConnection()->createQueryBuilder() |
| 34 | + ->select('id') |
| 35 | + ->from($this->getContext()->getTableName()) |
| 36 | + ->andWhere('queue IN (:queues)') |
| 37 | + ->andWhere('delayed_until IS NULL OR delayed_until <= :delayedUntil') |
| 38 | + ->andWhere('delivery_id IS NULL') |
| 39 | + ->addOrderBy('priority', 'asc') |
| 40 | + ->addOrderBy('published_at', 'asc') |
| 41 | + ->setParameter('queues', $queues, Connection::PARAM_STR_ARRAY) |
| 42 | + ->setParameter('delayedUntil', $now, ParameterType::INTEGER) |
| 43 | + ->setMaxResults(1); |
| 44 | + |
| 45 | + $update = $this->getConnection()->createQueryBuilder() |
| 46 | + ->update($this->getContext()->getTableName()) |
| 47 | + ->set('delivery_id', ':deliveryId') |
| 48 | + ->set('redeliver_after', ':redeliverAfter') |
| 49 | + ->andWhere('id = :messageId') |
| 50 | + ->andWhere('delivery_id IS NULL') |
| 51 | + ->setParameter('deliveryId', $deliveryId->getBytes(), Type::GUID) |
| 52 | + ->setParameter('redeliverAfter', $now + $redeliveryDelay, Type::BIGINT) |
| 53 | + ; |
45 | 54 |
|
| 55 | + while (microtime(true) < $endAt) { |
| 56 | + $result = $select->execute()->fetch(); |
| 57 | + if (empty($result)) { |
46 | 58 | return null; |
47 | 59 | } |
48 | 60 |
|
49 | | - // mark message as delivered to consumer |
50 | | - $this->getConnection()->createQueryBuilder() |
51 | | - ->andWhere('id = :id') |
52 | | - ->update($this->getContext()->getTableName()) |
53 | | - ->set('delivery_id', ':deliveryId') |
54 | | - ->set('redeliver_after', ':redeliverAfter') |
55 | | - ->setParameter('id', $message['id'], Type::GUID) |
56 | | - ->setParameter('deliveryId', $deliveryId, Type::STRING) |
57 | | - ->setParameter('redeliverAfter', $now + $redeliveryDelay, Type::BIGINT) |
58 | | - ->execute() |
| 61 | + $update |
| 62 | + ->setParameter('messageId', $result['id'], Type::GUID) |
59 | 63 | ; |
60 | 64 |
|
61 | | - $this->getConnection()->commit(); |
62 | | - |
63 | | - $deliveredMessage = $this->getConnection()->createQueryBuilder() |
64 | | - ->select('*') |
65 | | - ->from($this->getContext()->getTableName()) |
66 | | - ->andWhere('delivery_id = :deliveryId') |
67 | | - ->setParameter('deliveryId', $deliveryId, Type::STRING) |
68 | | - ->setMaxResults(1) |
69 | | - ->execute() |
70 | | - ->fetch() |
71 | | - ; |
72 | | - |
73 | | - return $deliveredMessage ?: null; |
74 | | - } catch (\Exception $e) { |
75 | | - $this->getConnection()->rollBack(); |
76 | | - |
77 | | - throw $e; |
| 65 | + if ($update->execute()) { |
| 66 | + $deliveredMessage = $this->getConnection()->createQueryBuilder() |
| 67 | + ->select('*') |
| 68 | + ->from($this->getContext()->getTableName()) |
| 69 | + ->andWhere('delivery_id = :deliveryId') |
| 70 | + ->setParameter('deliveryId', $deliveryId->getBytes(), Type::GUID) |
| 71 | + ->setMaxResults(1) |
| 72 | + ->execute() |
| 73 | + ->fetch() |
| 74 | + ; |
| 75 | + |
| 76 | + if (false == $deliveredMessage) { |
| 77 | + throw new \LogicException('There must be a message at all times at this stage but there is no a message.'); |
| 78 | + } |
| 79 | + |
| 80 | + if ($deliveredMessage['redelivered'] || empty($deliveredMessage['time_to_live']) || $deliveredMessage['time_to_live'] > time()) { |
| 81 | + return $this->getContext()->convertMessage($deliveredMessage); |
| 82 | + } |
| 83 | + } |
78 | 84 | } |
| 85 | + |
| 86 | + return null; |
79 | 87 | } |
80 | 88 |
|
81 | 89 | protected function redeliverMessages(): void |
82 | 90 | { |
83 | | - $this->getConnection()->createQueryBuilder() |
| 91 | + if (null === $this->redeliverMessagesLastExecutedAt) { |
| 92 | + $this->redeliverMessagesLastExecutedAt = microtime(true); |
| 93 | + } elseif ((microtime(true) - $this->redeliverMessagesLastExecutedAt) < 1) { |
| 94 | + return; |
| 95 | + } |
| 96 | + |
| 97 | + $update = $this->getConnection()->createQueryBuilder() |
84 | 98 | ->update($this->getContext()->getTableName()) |
85 | 99 | ->set('delivery_id', ':deliveryId') |
86 | 100 | ->set('redelivered', ':redelivered') |
87 | | - ->andWhere('delivery_id IS NOT NULL') |
88 | 101 | ->andWhere('redeliver_after < :now') |
89 | | - ->setParameter(':now', (int) time(), Type::BIGINT) |
90 | | - ->setParameter('deliveryId', null, Type::STRING) |
| 102 | + ->andWhere('delivery_id IS NOT NULL') |
| 103 | + ->setParameter(':now', time(), Type::BIGINT) |
| 104 | + ->setParameter('deliveryId', null, Type::GUID) |
91 | 105 | ->setParameter('redelivered', true, Type::BOOLEAN) |
92 | | - ->execute() |
93 | 106 | ; |
| 107 | + |
| 108 | + $update->execute(); |
| 109 | + |
| 110 | + $this->redeliverMessagesLastExecutedAt = microtime(true); |
94 | 111 | } |
95 | 112 |
|
96 | 113 | protected function removeExpiredMessages(): void |
97 | 114 | { |
98 | | - $this->getConnection()->createQueryBuilder() |
| 115 | + if (null === $this->removeExpiredMessagesLastExecutedAt) { |
| 116 | + $this->removeExpiredMessagesLastExecutedAt = microtime(true); |
| 117 | + } elseif ((microtime(true) - $this->removeExpiredMessagesLastExecutedAt) < 1) { |
| 118 | + return; |
| 119 | + } |
| 120 | + |
| 121 | + $delete = $this->getConnection()->createQueryBuilder() |
99 | 122 | ->delete($this->getContext()->getTableName()) |
100 | 123 | ->andWhere('(time_to_live IS NOT NULL) AND (time_to_live < :now)') |
101 | | - ->setParameter(':now', (int) time(), Type::BIGINT) |
102 | | - ->setParameter('redelivered', false, Type::BOOLEAN) |
103 | | - ->execute() |
| 124 | + ->andWhere('delivery_id IS NULL') |
| 125 | + ->andWhere('redelivered = false') |
| 126 | + |
| 127 | + ->setParameter(':now', time(), Type::BIGINT) |
104 | 128 | ; |
| 129 | + |
| 130 | + $delete->execute(); |
| 131 | + |
| 132 | + $this->removeExpiredMessagesLastExecutedAt = microtime(true); |
| 133 | + } |
| 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 | + ); |
105 | 146 | } |
106 | 147 | } |
0 commit comments