Skip to content
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

Jump to bottom

Amqp basic consume fixes #223

Merged
merged 6 commits into from
Oct 12, 2017
Merged

Amqp basic consume fixes #223

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
6 commits
Select commit Hold shift + click to select a range
b695dff
[amqp-bunny] fixes and impr related to basic consume feature.
makasim Oct 12, 2017
e4d894f
[amqp-lib] The context should allow to get the lib's channel.
makasim Oct 12, 2017
f4ca543
[amqp-lib] fixes and impr for basic consume feature.
makasim Oct 12, 2017
a8c288b
[amqp-ext] fixes and impr for basic consume feature.
makasim Oct 12, 2017
c2d1dab
[amqp-bunny] purge queues in specs.
makasim Oct 12, 2017
814443d
upd queue spec req
makasim Oct 12, 2017
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
113 changes: 20 additions & 93 deletions pkg/amqp-bunny/AmqpConsumer.php
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -3,18 +3,20 @@
namespace Enqueue\AmqpBunny;

use Bunny\Channel;
use Bunny\Client;
use Bunny\Message;
use Interop\Amqp\AmqpConsumer as InteropAmqpConsumer;
use Interop\Amqp\AmqpMessage as InteropAmqpMessage;
use Interop\Amqp\AmqpQueue as InteropAmqpQueue;
use Interop\Amqp\Impl\AmqpMessage;
use Interop\Queue\Exception;
use Interop\Queue\InvalidMessageException;
use Interop\Queue\PsrMessage;

class AmqpConsumer implements InteropAmqpConsumer
{
/**
* @var AmqpContext
*/
private $context;

/**
* @var Channel
*/
Expand All @@ -30,11 +32,6 @@ class AmqpConsumer implements InteropAmqpConsumer
*/
private $buffer;

/**
* @var bool
*/
private $isInit;

/**
* @var string
*/
Expand All @@ -51,36 +48,26 @@ class AmqpConsumer implements InteropAmqpConsumer
private $consumerTag;

/**
* @var Message
*/
private $bunnyMessages = [];

/**
* @param Channel $channel
* @param AmqpContext $context
* @param InteropAmqpQueue $queue
* @param Buffer $buffer
* @param string $receiveMethod
*/
public function __construct(Channel $channel, InteropAmqpQueue $queue, Buffer $buffer, $receiveMethod)
public function __construct(AmqpContext $context, InteropAmqpQueue $queue, Buffer $buffer, $receiveMethod)
{
$this->channel = $channel;
$this->context = $context;
$this->channel = $context->getBunnyChannel();
$this->queue = $queue;
$this->buffer = $buffer;
$this->receiveMethod = $receiveMethod;
$this->flags = self::FLAG_NOPARAM;

$this->isInit = false;
}

/**
* {@inheritdoc}
*/
public function setConsumerTag($consumerTag)
{
if ($this->isInit) {
throw new Exception('Consumer tag is not mutable after it has been subscribed to broker');
}

$this->consumerTag = $consumerTag;
}

Expand Down Expand Up @@ -154,9 +141,7 @@ public function receive($timeout = 0)
public function receiveNoWait()
{
if ($message = $this->channel->get($this->queue->getQueueName(), (bool) ($this->getFlags() & InteropAmqpConsumer::FLAG_NOACK))) {
$this->bunnyMessages[$message->deliveryTag] = $message;

return $this->convertMessage($message);
return $this->context->convertMessage($message);
}
}

Expand All @@ -167,11 +152,8 @@ public function acknowledge(PsrMessage $message)
{
InvalidMessageException::assertMessageInstanceOf($message, InteropAmqpMessage::class);

if (isset($this->bunnyMessages[$message->getDeliveryTag()])) {
$this->channel->ack($this->bunnyMessages[$message->getDeliveryTag()]);

unset($this->bunnyMessages[$message->getDeliveryTag()]);
}
$bunnyMessage = new Message('', $message->getDeliveryTag(), '', '', '', [], '');
$this->channel->ack($bunnyMessage);
}

/**
Expand All @@ -182,41 +164,8 @@ public function reject(PsrMessage $message, $requeue = false)
{
InvalidMessageException::assertMessageInstanceOf($message, InteropAmqpMessage::class);

if (isset($this->bunnyMessages[$message->getDeliveryTag()])) {
$this->channel->reject($this->bunnyMessages[$message->getDeliveryTag()], $requeue);

unset($this->bunnyMessages[$message->getDeliveryTag()]);
}
}

/**
* @param Message $bunnyMessage
*
* @return InteropAmqpMessage
*/
private function convertMessage(Message $bunnyMessage)
{
$headers = $bunnyMessage->headers;

$properties = [];
if (isset($headers['application_headers'])) {
$properties = $headers['application_headers'];
}
unset($headers['application_headers']);

if (array_key_exists('timestamp', $headers)) {
/** @var \DateTime $date */
$date = $headers['timestamp'];

$headers['timestamp'] = (int) $date->format('U');
}

$message = new AmqpMessage($bunnyMessage->content, $properties, $headers);
$message->setDeliveryTag($bunnyMessage->deliveryTag);
$message->setRedelivered($bunnyMessage->redelivered);
$message->setRoutingKey($bunnyMessage->routingKey);

return $message;
$bunnyMessage = new Message('', $message->getDeliveryTag(), '', '', '', [], '');
$this->channel->reject($bunnyMessage, $requeue);
}

/**
Expand Down Expand Up @@ -244,34 +193,12 @@ private function receiveBasicGet($timeout)
*/
private function receiveBasicConsume($timeout)
{
if (false === $this->isInit) {
$callback = function (Message $message, Channel $channel, Client $bunny) {
$receivedMessage = $this->convertMessage($message);
$receivedMessage->setConsumerTag($message->consumerTag);

$this->bunnyMessages[$message->deliveryTag] = $message;
$this->buffer->push($receivedMessage->getConsumerTag(), $receivedMessage);

$bunny->stop();
};

$frame = $this->channel->consume(
$callback,
$this->queue->getQueueName(),
$this->getConsumerTag() ?: $this->getQueue()->getConsumerTag(),
(bool) ($this->getFlags() & InteropAmqpConsumer::FLAG_NOLOCAL),
(bool) ($this->getFlags() & InteropAmqpConsumer::FLAG_NOACK),
(bool) ($this->getFlags() & InteropAmqpConsumer::FLAG_EXCLUSIVE),
(bool) ($this->getFlags() & InteropAmqpConsumer::FLAG_NOWAIT)
);

$this->consumerTag = $frame->consumerTag;

if (empty($this->consumerTag)) {
throw new Exception('Got empty consumer tag');
}
if (false == $this->consumerTag) {
$this->context->subscribe($this, function (InteropAmqpMessage $message) {
$this->buffer->push($message->getConsumerTag(), $message);

$this->isInit = true;
return false;
});
}

if ($message = $this->buffer->pop($this->consumerTag)) {
Expand All @@ -281,7 +208,7 @@ private function receiveBasicConsume($timeout)
while (true) {
$start = microtime(true);

$this->channel->getClient()->run($timeout / 1000);
$this->context->consume($timeout);

if ($message = $this->buffer->pop($this->consumerTag)) {
return $message;
Expand Down
10 changes: 6 additions & 4 deletions pkg/amqp-bunny/AmqpContext.php
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -127,10 +127,10 @@ public function createConsumer(PsrDestination $destination)
$queue = $this->createTemporaryQueue();
$this->bind(new AmqpBind($destination, $queue, $queue->getQueueName()));

return new AmqpConsumer($this->getBunnyChannel(), $queue, $this->buffer, $this->config['receive_method']);
return new AmqpConsumer($this, $queue, $this->buffer, $this->config['receive_method']);
}

return new AmqpConsumer($this->getBunnyChannel(), $destination, $this->buffer, $this->config['receive_method']);
return new AmqpConsumer($this, $destination, $this->buffer, $this->config['receive_method']);
}

/**
Expand Down Expand Up @@ -411,11 +411,13 @@ public function getBunnyChannel()
}

/**
* @internal It must be used here and in the consumer only
*
* @param Message $bunnyMessage
*
* @return InteropAmqpMessage
*/
private function convertMessage(Message $bunnyMessage)
public function convertMessage(Message $bunnyMessage)
{
$headers = $bunnyMessage->headers;

Expand All @@ -425,7 +427,7 @@ private function convertMessage(Message $bunnyMessage)
}
unset($headers['application_headers']);

if (array_key_exists('timestamp', $headers)) {
if (array_key_exists('timestamp', $headers) && $headers['timestamp']) {
/** @var \DateTime $date */
$date = $headers['timestamp'];

Expand Down
2 changes: 1 addition & 1 deletion pkg/amqp-bunny/AmqpProducer.php
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -77,7 +77,7 @@ public function send(PsrDestination $destination, PsrMessage $message)

$amqpProperties = $message->getHeaders();

if (array_key_exists('timestamp', $amqpProperties)) {
if (array_key_exists('timestamp', $amqpProperties) && null !== $amqpProperties['timestamp']) {
$amqpProperties['timestamp'] = \DateTime::createFromFormat('U', $amqpProperties['timestamp']);
}

Expand Down
Loading