Skip to content

[BC break] Amqp add basic consume support #217

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.

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

Merged
merged 12 commits into from
Oct 11, 2017
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
4 changes: 2 additions & 2 deletions composer.json
Original file line number Diff line number Diff line change
Expand Up @@ -27,8 +27,8 @@
"enqueue/test": "*@dev",
"enqueue/async-event-dispatcher": "*@dev",
"queue-interop/queue-interop": "^0.6@dev",
"queue-interop/amqp-interop": "^0.6@dev",
"queue-interop/queue-spec": "^0.5@dev",
"queue-interop/amqp-interop": "^0.7@dev",
"queue-interop/queue-spec": "^0.5.1@dev",

"phpunit/phpunit": "^5",
"doctrine/doctrine-bundle": "~1.2",
Expand Down
1 change: 1 addition & 0 deletions docker/Dockerfile
Original file line number Diff line number Diff line change
Expand Up @@ -17,6 +17,7 @@ RUN echo "extension=rdkafka.so" > /etc/php/7.1/cli/conf.d/10-rdkafka.ini
RUN echo "extension=rdkafka.so" > /etc/php/7.1/fpm/conf.d/10-rdkafka.ini

COPY ./php/cli.ini /etc/php/7.1/cli/conf.d/1-dev_cli.ini
COPY ./php/amqp.so /usr/lib/php/20160303/amqp.so
COPY ./bin/dev_entrypoiny.sh /usr/local/bin/entrypoint.sh
RUN chmod u+x /usr/local/bin/entrypoint.sh

Expand Down
Binary file added docker/php/amqp.so
Binary file not shown.
3 changes: 2 additions & 1 deletion phpstan.neon
Original file line number Diff line number Diff line change
Expand Up @@ -7,4 +7,5 @@ parameters:
- pkg/redis/PhpRedis.php
- pkg/redis/RedisConnectionFactory.php
- pkg/gearman
- pkg/amqp-ext/AmqpConsumer.php
- pkg/amqp-ext/AmqpConsumer.php
- pkg/amqp-ext/AmqpContext.php
104 changes: 104 additions & 0 deletions pkg/amqp-bunny/AmqpContext.php
Original file line number Diff line number Diff line change
Expand Up @@ -3,9 +3,12 @@
namespace Enqueue\AmqpBunny;

use Bunny\Channel;
use Bunny\Client;
use Bunny\Message;
use Enqueue\AmqpTools\DelayStrategyAware;
use Enqueue\AmqpTools\DelayStrategyAwareTrait;
use Interop\Amqp\AmqpBind as InteropAmqpBind;
use Interop\Amqp\AmqpConsumer as InteropAmqpConsumer;
use Interop\Amqp\AmqpContext as InteropAmqpContext;
use Interop\Amqp\AmqpMessage as InteropAmqpMessage;
use Interop\Amqp\AmqpQueue as InteropAmqpQueue;
Expand Down Expand Up @@ -43,6 +46,13 @@ class AmqpContext implements InteropAmqpContext, DelayStrategyAware
*/
private $buffer;

/**
* an item contains an array: [AmqpConsumerInterop $consumer, callable $callback];.
*
* @var array
*/
private $subscribers;

/**
* Callable must return instance of \Bunny\Channel once called.
*
Expand Down Expand Up @@ -309,6 +319,77 @@ public function setQos($prefetchSize, $prefetchCount, $global)
$this->getBunnyChannel()->qos($prefetchSize, $prefetchCount, $global);
}

/**
* {@inheritdoc}
*/
public function subscribe(InteropAmqpConsumer $consumer, callable $callback)
{
if ($consumer->getConsumerTag() && array_key_exists($consumer->getConsumerTag(), $this->subscribers)) {
return;
}

$bunnyCallback = function (Message $message, Channel $channel, Client $bunny) {
$receivedMessage = $this->convertMessage($message);
$receivedMessage->setConsumerTag($message->consumerTag);

/**
* @var AmqpConsumer
* @var callable $callback
*/
list($consumer, $callback) = $this->subscribers[$message->consumerTag];

if (false === call_user_func($callback, $receivedMessage, $consumer)) {
$bunny->stop();
}
};

$frame = $this->getBunnyChannel()->consume(
$bunnyCallback,
$consumer->getQueue()->getQueueName(),
$consumer->getConsumerTag(),
(bool) ($consumer->getFlags() & InteropAmqpConsumer::FLAG_NOLOCAL),
(bool) ($consumer->getFlags() & InteropAmqpConsumer::FLAG_NOACK),
(bool) ($consumer->getFlags() & InteropAmqpConsumer::FLAG_EXCLUSIVE),
(bool) ($consumer->getFlags() & InteropAmqpConsumer::FLAG_NOWAIT)
);

if (empty($frame->consumerTag)) {
throw new Exception('Got empty consumer tag');
}

$consumer->setConsumerTag($frame->consumerTag);

$this->subscribers[$frame->consumerTag] = [$consumer, $callback];
}

/**
* {@inheritdoc}
*/
public function unsubscribe(InteropAmqpConsumer $consumer)
{
if (false == $consumer->getConsumerTag()) {
return;
}

$consumerTag = $consumer->getConsumerTag();

$this->getBunnyChannel()->cancel($consumerTag);
$consumer->setConsumerTag(null);
unset($this->subscribers[$consumerTag]);
}

/**
* {@inheritdoc}
*/
public function consume($timeout = 0)
{
if (empty($this->subscribers)) {
throw new \LogicException('There is no subscribers. Consider calling basicConsumeSubscribe before consuming');
}

$this->getBunnyChannel()->getClient()->run($timeout / 1000);
}

/**
* @return Channel
*/
Expand All @@ -328,4 +409,27 @@ public function getBunnyChannel()

return $this->bunnyChannel;
}

/**
* @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']);

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

return $message;
}
}
22 changes: 22 additions & 0 deletions pkg/amqp-bunny/Tests/Spec/AmqpBasicConsumeBreakOnFalseTest.php
Original file line number Diff line number Diff line change
@@ -0,0 +1,22 @@
<?php

namespace Enqueue\AmqpBunny\Tests\Spec;

use Enqueue\AmqpBunny\AmqpConnectionFactory;
use Interop\Queue\Spec\Amqp\BasicConsumeBreakOnFalseSpec;

/**
* @group functional
*/
class AmqpBasicConsumeBreakOnFalseTest extends BasicConsumeBreakOnFalseSpec
{
/**
* {@inheritdoc}
*/
protected function createContext()
{
$factory = new AmqpConnectionFactory(getenv('AMQP_DSN'));

return $factory->createContext();
}
}
Original file line number Diff line number Diff line change
@@ -0,0 +1,22 @@
<?php

namespace Enqueue\AmqpBunny\Tests\Spec;

use Enqueue\AmqpBunny\AmqpConnectionFactory;
use Interop\Queue\Spec\Amqp\BasicConsumeFromAllSubscribedQueuesSpec;

/**
* @group functional
*/
class AmqpBasicConsumeFromAllSubscribedQueuesTest extends BasicConsumeFromAllSubscribedQueuesSpec
{
/**
* {@inheritdoc}
*/
protected function createContext()
{
$factory = new AmqpConnectionFactory(getenv('AMQP_DSN'));

return $factory->createContext();
}
}
Original file line number Diff line number Diff line change
@@ -0,0 +1,22 @@
<?php

namespace Enqueue\AmqpBunny\Tests\Spec;

use Enqueue\AmqpBunny\AmqpConnectionFactory;
use Interop\Queue\Spec\Amqp\BasicConsumeShouldAddConsumerTagOnSubscribeSpec;

/**
* @group functional
*/
class AmqpBasicConsumeShouldAddConsumerTagOnSubscribeTest extends BasicConsumeShouldAddConsumerTagOnSubscribeSpec
{
/**
* {@inheritdoc}
*/
protected function createContext()
{
$factory = new AmqpConnectionFactory(getenv('AMQP_DSN'));

return $factory->createContext();
}
}
Original file line number Diff line number Diff line change
@@ -0,0 +1,22 @@
<?php

namespace Enqueue\AmqpBunny\Tests\Spec;

use Enqueue\AmqpBunny\AmqpConnectionFactory;
use Interop\Queue\Spec\Amqp\BasicConsumeShouldRemoveConsumerTagOnUnsubscribeSpec;

/**
* @group functional
*/
class AmqpBasicConsumeShouldRemoveConsumerTagOnUnsubscribeTest extends BasicConsumeShouldRemoveConsumerTagOnUnsubscribeSpec
{
/**
* {@inheritdoc}
*/
protected function createContext()
{
$factory = new AmqpConnectionFactory(getenv('AMQP_DSN'));

return $factory->createContext();
}
}
Original file line number Diff line number Diff line change
@@ -0,0 +1,22 @@
<?php

namespace Enqueue\AmqpBunny\Tests\Spec;

use Enqueue\AmqpBunny\AmqpConnectionFactory;
use Interop\Queue\Spec\Amqp\BasicConsumeUntilUnsubscribedSpec;

/**
* @group functional
*/
class AmqpBasicConsumeUntilUnsubscribedTest extends BasicConsumeUntilUnsubscribedSpec
{
/**
* {@inheritdoc}
*/
protected function createContext()
{
$factory = new AmqpConnectionFactory(getenv('AMQP_DSN'));

return $factory->createContext();
}
}
4 changes: 2 additions & 2 deletions pkg/amqp-bunny/composer.json
Original file line number Diff line number Diff line change
Expand Up @@ -7,7 +7,7 @@
"require": {
"php": ">=5.6",

"queue-interop/amqp-interop": "^0.6@dev",
"queue-interop/amqp-interop": "^0.7@dev",
"bunny/bunny": "^0.2.4",
"enqueue/amqp-tools": "^0.8@dev"
},
Expand All @@ -16,7 +16,7 @@
"enqueue/test": "^0.8@dev",
"enqueue/enqueue": "^0.8@dev",
"enqueue/null": "^0.8@dev",
"queue-interop/queue-spec": "^0.5@dev",
"queue-interop/queue-spec": "^0.5.1@dev",
"symfony/dependency-injection": "^2.8|^3",
"symfony/config": "^2.8|^3"
},
Expand Down
Loading