Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
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
7 changes: 1 addition & 6 deletions src/Queue/Jobs/RabbitMQJob.php
Original file line number Diff line number Diff line change
Expand Up @@ -107,7 +107,6 @@ public function release($delay = 0)
parent::release($delay);

$this->delete();
$this->connection->setAttempts($this->attempts() + 1);

$body = $this->payload();

Expand All @@ -122,11 +121,7 @@ public function release($delay = 0)

$data = $body['data'];

if ($delay > 0) {
$this->connection->later($delay, $job, $data, $this->getQueue());
} else {
$this->connection->push($job, $data, $this->getQueue());
}
$this->connection->release($delay, $job, $data, $this->getQueue(), $this->attempts() + 1);
}

/**
Expand Down
40 changes: 20 additions & 20 deletions src/Queue/RabbitMQQueue.php
Original file line number Diff line number Diff line change
Expand Up @@ -15,11 +15,6 @@

class RabbitMQQueue extends Queue implements QueueContract
{
/**
* Used for retry logic, to set the retries on the message metadata instead of the message body.
*/
const ATTEMPT_COUNT_HEADERS_KEY = 'attempts_count';

protected $sleepOnError;

protected $queueOptions;
Expand All @@ -32,7 +27,6 @@ class RabbitMQQueue extends Queue implements QueueContract
* @var AmqpContext
*/
private $context;
private $retryAfter;
private $correlationId;

public function __construct(AmqpContext $context, array $config)
Expand Down Expand Up @@ -78,8 +72,8 @@ public function pushRaw($payload, $queueName = null, array $options = [])
$message->setContentType('application/json');
$message->setDeliveryMode(AmqpMessage::DELIVERY_MODE_PERSISTENT);

if ($this->retryAfter !== null) {
$message->setProperty(self::ATTEMPT_COUNT_HEADERS_KEY, $this->retryAfter);
if (isset($options['attempts'])) {
$message->setProperty(RabbitMQJob::ATTEMPT_COUNT_HEADERS_KEY, $options['attempts']);
}

$producer = $this->context->createProducer();
Expand All @@ -103,6 +97,24 @@ public function later($delay, $job, $data = '', $queue = null)
return $this->pushRaw($this->createPayload($job, $data), $queue, ['delay' => $this->secondsUntil($delay)]);
}

/**
* Release a reserved job back onto the queue.
*
* @param \DateTimeInterface|\DateInterval|int $delay
* @param string|object $job
* @param mixed $data
* @param string $queue
* @param int $attempts
* @return mixed
*/
public function release($delay, $job, $data, $queue, $attempts = 0)
{
return $this->pushRaw($this->createPayload($job, $data), $queue, [
'delay' => $this->secondsUntil($delay),
'attempts' => $attempts
]);
}

/** @inheritdoc */
public function pop($queueName = null)
{
Expand All @@ -122,18 +134,6 @@ public function pop($queueName = null)
return null;
}

/**
* Sets the attempts member variable to be used in message generation.
*
* @param int $count
*
* @return void
*/
public function setAttempts(int $count)
{
$this->retryAfter = $count;
}

/**
* Retrieves the correlation id, or a unique id.
*
Expand Down
18 changes: 4 additions & 14 deletions tests/Queue/RabbitMQQueueTest.php
Original file line number Diff line number Diff line change
Expand Up @@ -34,15 +34,6 @@ public function testCouldBeConstructedWithExpectedArguments()
new RabbitMQQueue($this->createAmqpContext(), $this->createDummyConfig());
}

public function testShouldAllowSetAttempts()
{
$queue = new RabbitMQQueue($this->createAmqpContext(), $this->createDummyConfig());

$queue->setAttempts(123);

$this->assertAttributeSame(123, 'retryAfter', $queue);
}

public function testShouldGenerateNewCorrelationIdIfNotSet()
{
$queue = new RabbitMQQueue($this->createAmqpContext(), $this->createDummyConfig());
Expand Down Expand Up @@ -128,7 +119,7 @@ public function testShouldSendExpectedMessageOnPushRaw()
$this->assertSame('application/json', $message->getContentType());
$this->assertSame(AmqpMessage::DELIVERY_MODE_PERSISTENT, $message->getDeliveryMode());
$this->assertNotEmpty($message->getCorrelationId());
$this->assertNull($message->getProperty(RabbitMQQueue::ATTEMPT_COUNT_HEADERS_KEY));
$this->assertNull($message->getProperty(RabbitMQJob::ATTEMPT_COUNT_HEADERS_KEY));
})
;
$producer
Expand Down Expand Up @@ -167,7 +158,7 @@ public function testShouldSendExpectedMessageOnPushRaw()
$queue->pushRaw('thePayload', $expectedQueueName);
}

public function testShouldSetAttemptCountHeaderIfNotNull()
public function testShouldSetAttemptCountPropIfNotNull()
{
$expectedAttempts = 54321;

Expand All @@ -179,7 +170,7 @@ public function testShouldSetAttemptCountHeaderIfNotNull()
->method('send')
->with($this->identicalTo($topic), $this->isInstanceOf(AmqpMessage::class))
->willReturnCallback(function ($actualTopic, AmqpMessage $message) use ($expectedAttempts) {
$this->assertSame($expectedAttempts, $message->getProperty(RabbitMQQueue::ATTEMPT_COUNT_HEADERS_KEY));
$this->assertSame($expectedAttempts, $message->getProperty(RabbitMQJob::ATTEMPT_COUNT_HEADERS_KEY));
})
;
$producer
Expand Down Expand Up @@ -212,9 +203,8 @@ public function testShouldSetAttemptCountHeaderIfNotNull()

$queue = new RabbitMQQueue($context, $this->createDummyConfig());
$queue->setContainer($this->createDummyContainer());
$queue->setAttempts($expectedAttempts);

$queue->pushRaw('thePayload', 'aQueue');
$queue->pushRaw('thePayload', 'aQueue', ['attempts' => $expectedAttempts]);
}

public function testShouldSetDeliveryDelayIfDelayOptionPresent()
Expand Down