Skip to content

Allow to disable non-blocking mode #296

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
wants to merge 3 commits into from
Closed

Allow to disable non-blocking mode #296

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
4bced67
Update Consumer.php
MorrisonHotel Dec 30, 2019
aca26a2
Update ConsumeCommand.php
MorrisonHotel Dec 30, 2019
d182d03
Update ConsumeCommand.php
MorrisonHotel Dec 30, 2019
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 2 additions & 0 deletions src/Console/ConsumeCommand.php
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -23,6 +23,7 @@ class ConsumeCommand extends WorkCommand
{--consumer-tag}
{--prefetch-size=0}
{--prefetch-count=1000}
{--blocking : Use blocking mode to reduce CPU usage}
';

protected $description = 'Consume messages';
Expand All @@ -36,6 +37,7 @@ public function handle(): void
$consumer->setConsumerTag($this->consumerTag());
$consumer->setPrefetchSize((int) $this->option('prefetch-size'));
$consumer->setPrefetchCount((int) $this->option('prefetch-count'));
$consumer->setNonBlocking(! $this->option('blocking'));

parent::handle();
}
Expand Down
10 changes: 9 additions & 1 deletion src/Consumer.php
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -32,6 +32,9 @@ class Consumer extends Worker
/** @var AMQPChannel */
protected $channel;

/** @var bool */
protected $nonBlocking = true;

public function setContainer(Container $value): void
{
$this->container = $value;
Expand All @@ -52,6 +55,11 @@ public function setPrefetchCount(int $value): void
$this->prefetchCount = $value;
}

public function setNonBlocking(bool $nonBlocking): void
{
$this->nonBlocking = $nonBlocking;
}

public function daemon($connectionName, $queue, WorkerOptions $options): void
{
if ($this->supportsAsyncSignals()) {
Expand Down Expand Up @@ -108,7 +116,7 @@ function (AMQPMessage $message) use ($connection, $options, $connectionName, $qu
// fire off this job for processing. Otherwise, we will need to sleep the
// worker so no more jobs are processed until they should be processed.
try {
$this->channel->wait(null, true, (int) $options->timeout);
$this->channel->wait(null, $this->nonBlocking, (int) $options->timeout);
} catch (AMQPRuntimeException $exception) {
$this->exceptions->report($exception);

Expand Down