diogo/symfony
Archived
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

bug #31396 [Messenger] Fix rejecting of pending messages (alexander-schranz)

Browse Source 
This PR was squashed before being merged into the 4.3-dev branch (closes #31396).

Discussion
----------

[Messenger] Fix rejecting of pending messages

| Q             | A
| ------------- | ---
| Branch?       | master
| Bug fix?      | yes
| New feature?  | no
| BC breaks?    | no
| Deprecations? | no
| Tests pass?   | yes
| Fixed tickets | #...
| License       | MIT
| Doc PR        | -

As discussed in https://github.com/symfony/symfony/pull/31387#issuecomment-489605683. It seems pending messages are not removed by calling xdel so we need to call xack first to remove them from the pending state and then call xdel to remove it from the complete stream.

It seems to be the correct way as mentioned in: https://github.com/antirez/redis/issues/5754

Test should be the same as added in: https://github.com/symfony/symfony/pull/31387/files#diff-46c1e03dafbcebc46b5cace7d05de20c

Commits
-------

072e466355 [Messenger] Fix rejecting of pending messages
This commit is contained in:
Fabien Potencier 2019-05-06 17:19:52 +02:00
commit bec45edbea
1 changed files with 3 additions and 2 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

5
src/Symfony/Component/Messenger/Transport/RedisExt/Connection.php
View File

@ -89,7 +89,7 @@ class Connection
} catch (\RedisException $e) {
}
if ($e || (false === $messages && !$this->couldHavePendingMessages)) {
if ($e || false === $messages) {
throw new TransportException(
($e ? $e->getMessage() : $this->connection->getLastError()) ?? 'Could not read messages from the redis stream.'
);
@ -132,7 +132,8 @@ class Connection
{
$e = null;
try {
$deleted = $this->connection->xdel($this->stream, [$id]);
$deleted = $this->connection->xack($this->stream, $this->group, [$id]);
$deleted = $this->connection->xdel($this->stream, [$id]) && $deleted;
} catch (\RedisException $e) {
}