nsq/nsqphp
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

Refactoring: Simplify Message methods, add isProcessed method.

Browse Source
This commit is contained in:
Konstantin Grachev
2021-06-09 16:33:18 +03:00
parent 2fc7e37120
commit d3e1788d23
1 changed files with 19 additions and 24 deletions
Download Patch File Download Diff File Expand all files Collapse all files

43
src/Message.php
View File

@ -6,7 +6,6 @@ namespace Nsq;
use Amp\Promise;
use Nsq\Exception\MessageException;
use function Amp\call;
final class Message
{
@ -32,20 +31,19 @@ final class Message
);
}
public function isProcessed(): bool
{
return $this->processed;
}
/**
* @return Promise<void>
*/
public function finish(): Promise
{
return call(function (): \Generator {
if ($this->processed) {
throw MessageException::processed($this);
}
$this->markAsProcessedOrFail();
yield $this->consumer->fin($this->id);
$this->processed = true;
});
return $this->consumer->fin($this->id);
}
/**
@ -53,15 +51,9 @@ final class Message
*/
public function requeue(int $timeout): Promise
{
return call(function () use ($timeout): \Generator {
if ($this->processed) {
throw MessageException::processed($this);
}
$this->markAsProcessedOrFail();
yield $this->consumer->req($this->id, $timeout);
$this->processed = true;
});
return $this->consumer->req($this->id, $timeout);
}
/**
@ -69,14 +61,17 @@ final class Message
*/
public function touch(): Promise
{
return call(function (): \Generator {
if ($this->processed) {
throw MessageException::processed($this);
}
$this->markAsProcessedOrFail();
yield $this->consumer->touch($this->id);
return $this->consumer->touch($this->id);
}
$this->processed = true;
});
private function markAsProcessedOrFail(): void
{
if ($this->processed) {
throw MessageException::processed($this);
}
$this->processed = true;
}
}