[Messenger] Fix rejecting of pending messages

alexander-schranz · fabpot · commit 072e46635590 · 2019-05-06T17:19:44.000+02:00
diff --git a/src/Symfony/Component/Messenger/Transport/RedisExt/Connection.php b/src/Symfony/Component/Messenger/Transport/RedisExt/Connection.php
@@ -89,7 +89,7 @@ public function get(): ?array
         } 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 @@ public function reject(string $id): void
     {
         $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) {
         }
 

Original file line number	Diff line number	Diff line change
`@@ -89,7 +89,7 @@ public function get(): ?array`
`89`	`89`	`} catch (\RedisException $e) {`
`90`	`90`	`}`
`91`	`91`
`92`		`- if ($e \|\| (false === $messages && !$this->couldHavePendingMessages)) {`
	`92`	`+ if ($e \|\| false === $messages) {`
`93`	`93`	`throw new TransportException(`
`94`	`94`	`($e ? $e->getMessage() : $this->connection->getLastError()) ?? 'Could not read messages from the redis stream.'`
`95`	`95`	`);`
`@@ -132,7 +132,8 @@ public function reject(string $id): void`
`132`	`132`	`{`
`133`	`133`	`$e = null;`
`134`	`134`	`try {`
`135`		`- $deleted = $this->connection->xdel($this->stream, [$id]);`
	`135`	`+ $deleted = $this->connection->xack($this->stream, $this->group, [$id]);`
	`136`	`+ $deleted = $this->connection->xdel($this->stream, [$id]) && $deleted;`
`136`	`137`	`} catch (\RedisException $e) {`
`137`	`138`	`}`
`138`	`139`