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.

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

Add AMQP interop based driver. #158

Merged
merged 17 commits into from Dec 5, 2017

Conversation

Projects
None yet
4 participants
@makasim
Copy link
Contributor

commented Oct 25, 2017

Advantages:

@makasim

This comment has been minimized.

Copy link
Contributor Author

commented Oct 25, 2017

I'll work on tests and other stuff once we agree in general on this matter.

@makasim

This comment has been minimized.

Copy link
Contributor Author

commented Oct 26, 2017

@samdark @zhuravljov could you please have a look?

@samdark

This comment has been minimized.

Copy link
Member

commented Oct 26, 2017

I don't have enough experience do decide on this one in general so @zhuravljov this one is yours. I'll check code in general.

$this->context->createProducer()
// TODO check values compatibility

This comment has been minimized.

Copy link
@samdark

samdark Oct 26, 2017

Member

TODO!

This comment has been minimized.

Copy link
@makasim

makasim Oct 26, 2017

Author Contributor

I'll work on tests and other stuff once we agree in general on this matter.

$this->channel->queue_declare($this->queueName, false, true, false, false);
$this->channel->exchange_declare($this->exchangeName, 'direct', false, true, false);
$this->channel->queue_bind($this->queueName, $this->exchangeName);
if ($this->context) return;

This comment has been minimized.

Copy link
@samdark

samdark Oct 26, 2017

Member

Add { and } please.

This comment has been minimized.

Copy link
@makasim

makasim Oct 26, 2017

Author Contributor

sure, (that's how it was, I kept original styling).

if (!$this->channel) return;
$this->channel->close();
$this->connection->close();
if (!$this->context) return;

This comment has been minimized.

Copy link
@samdark

samdark Oct 26, 2017

Member

Add { and } please.

@makasim makasim referenced this pull request Oct 26, 2017

Open

Proposals #4

@makasim

This comment has been minimized.

Copy link
Contributor Author

commented Oct 26, 2017

This implementation also supports new features: ttr, attempts, delays, priorities.

@makasim

This comment has been minimized.

Copy link
Contributor Author

commented Oct 30, 2017

@zhuravljov I was wondering if you have you got time to look into this PR?

'host' => $this->host,
'port' => $this->port,
'user' => $this->user,
'pass' => $this->password,

This comment has been minimized.

Copy link
@thiagotalma

thiagotalma Oct 31, 2017

Contributor

@makasim I just sent #162 to allow the configuration of vhost. You can include in this PR as well, it is a necessary configuration.

This comment has been minimized.

Copy link
@makasim

makasim Oct 31, 2017

Author Contributor

sure thing

@zhuravljov

This comment has been minimized.

Copy link
Member

commented Nov 1, 2017

@makasim thank you for contribution, I will look and test soon.

@makasim

This comment has been minimized.

Copy link
Contributor Author

commented Nov 1, 2017

Added:

  • Contains new options like: vhost, connection_timeout, qos_prefetch_count and so on.
  • Supports Secure (SSL) AMQP connections.
  • Add ability to set DSN like: amqp:, amqps: or amqp://user:pass@localhost:1000/vhost
@makasim

This comment has been minimized.

Copy link
Contributor Author

commented Nov 15, 2017

Would it be easier for you to review it if I, instead of changing existing class, create new ones?

@makasim makasim changed the title Amqp interop based driver. Add AMQ interop based driver. Nov 17, 2017

@makasim makasim force-pushed the formapro-forks:amqp-interop branch 3 times, most recently from e5eea13 to 44570a3 Nov 17, 2017

@makasim makasim force-pushed the formapro-forks:amqp-interop branch from bfe426f to ba7a8b4 Nov 17, 2017

@makasim

This comment has been minimized.

Copy link
Contributor Author

commented Nov 19, 2017

A test fail is not related to my changes.

makasim added some commits Nov 29, 2017

@makasim

This comment has been minimized.

Copy link
Contributor Author

commented Nov 29, 2017

@samdark @zhuravljov @thiagotalma I addressed all review comments.

@@ -4,8 +4,7 @@ Yii2 Queue Extension Change Log
2.0.2 under development
-----------------------

- no changes in this release.

- Enh: Add Amqp Interop driver (makasim)

This comment has been minimized.

Copy link
@samdark
@@ -18,6 +18,7 @@ Queue Drivers
* [Db](driver-db.md)
* [Redis](driver-redis.md)
* [RabbitMQ](driver-amqp.md)
* [AMQP Interop)](driver-amqp-interop.md)

This comment has been minimized.

Copy link
@samdark

samdark Nov 29, 2017

Member

Please remove extra ).


* It would work with any amqp interop compatible transports, such as

* [enqueue/amqp-ext](https://github.com/php-enqueue/amqp-ext) based on [php amqp extension](https://github.com/pdezwart/php-amqp)

This comment has been minimized.

Copy link
@samdark

samdark Nov 29, 2017

Member

php → PHP

/**
* Manages application amqp-queue.
*
* @author Maksym Kotliar <kotlyar.maksim@gmail.com>

This comment has been minimized.

Copy link
@samdark

samdark Nov 29, 2017

Member

@since 2.0.2 (or which version is it going to be released it, @zhuravljov ?)

*/
class Queue extends CliQueue
{
const H_ATTEMPT = 'yii-attempt';

This comment has been minimized.

Copy link
@samdark

samdark Nov 29, 2017

Member

What is H_? Do we need it?

This comment has been minimized.

Copy link
@makasim

makasim Nov 29, 2017

Author Contributor

H stands for Head.

This comment has been minimized.

Copy link
@samdark

samdark Nov 29, 2017

Member

Head of what?

$connectionClass = AmqpLibConnectionFactory::class;
break;
case 'ext':

This comment has been minimized.

Copy link
@samdark

samdark Nov 29, 2017

Member

Move to class constants.

$connectionClass = AmqpExtConnectionFactory::class;
break;
case 'bunny':

This comment has been minimized.

Copy link
@samdark

samdark Nov 29, 2017

Member

Move to class constants.

break;
default:
throw new \LogicException(sprintf('The given driver "%s" is not supported. Supported are "%s"', $this->driver, implode('", "', $this->supportedDrivers)));

This comment has been minimized.

Copy link
@samdark

samdark Nov 29, 2017

Member

Supported are "%s" → Drivers supported are "%s"

{
$queue = $this->context->createQueue($this->queueName);
$queue->addFlag(AmqpQueue::FLAG_DURABLE);
$queue->setArguments(['x-max-priority' => 4]);

This comment has been minimized.

Copy link
@samdark

samdark Nov 29, 2017

Member

What is 4? Why 4?

This comment has been minimized.

Copy link
@makasim

makasim Nov 29, 2017

Author Contributor

You can set a priority between 0 and 4. Here's the priorities doc

Resource Usage Considerations

If priority queues are desired, we recommend using between 1 and 10. Currently using more priorities will consume more resources (Erlang processes).
There is some in-memory and on-disk cost per priority level per queue. There is also an additional CPU cost, especially when consuming, so you may not wish to create huge numbers of levels.
The message priority field is defined as an unsigned byte, so in practice priorities should be between 0 and 255.

I add an ability to configure it and set the default value to 10 (as suggested by off doc).

Removed: Added an exception for the case where a developer wants to use a higher value. because of

Messages with a priority which is higher than the queue's maximum are treated as if they were published with the maximum priority.
public function testRun()
{
// Not supported

This comment has been minimized.

Copy link
@samdark

samdark Nov 29, 2017

Member

markTestSkipped()?

@makasim

This comment has been minimized.

Copy link
Contributor Author

commented Nov 29, 2017

@samdark done

*/
class Queue extends CliQueue
{
const H_ATTEMPT = 'yii-attempt';

This comment has been minimized.

Copy link
@samdark

samdark Nov 29, 2017

Member

I think we need to remove H_.

This comment has been minimized.

Copy link
@makasim

makasim Nov 29, 2017

Author Contributor

done

@makasim

This comment has been minimized.

Copy link
Contributor Author

commented Nov 29, 2017

Travis build fail is not related to my changes

@makasim makasim changed the title Add AMQ interop based driver. Add AMQP interop based driver. Nov 30, 2017

fix
@makasim

This comment has been minimized.

Copy link
Contributor Author

commented Nov 30, 2017

@zhuravljov could you please give a hand with the error on Travis?

@makasim

This comment has been minimized.

Copy link
Contributor Author

commented Nov 30, 2017

@zhuravljov tests failed because of this commit d456b75.

While fixing it I realized that it is a good idea to keep setup broker logic and connection one separate.

If you mix them up you won't be able to do anything if setup broker fails. For example, the broker has the same queue but with a different argument for x-max-priorities and hence it complains. You won't able to do anything from the code, like deleting the queue and creating it from scratch.

@zhuravljov zhuravljov merged commit b4669cc into yiisoft:master Dec 5, 2017

1 check passed

continuous-integration/travis-ci/pr The Travis CI build passed
Details
@zhuravljov

This comment has been minimized.

Copy link
Member

commented Dec 5, 2017

It is merged. Thank you for big work!

@zhuravljov zhuravljov referenced this pull request Dec 5, 2017

Closed

Queue Interop Support #104

@makasim makasim deleted the formapro-forks:amqp-interop branch Dec 5, 2017

@makasim

This comment has been minimized.

Copy link
Contributor Author

commented Dec 5, 2017

Party time! 🎉 🎉 🎉

Thank you guys for your patience and thoughtful reviews!

zhuravljov added a commit that referenced this pull request Dec 6, 2017

zhuravljov added a commit that referenced this pull request Dec 24, 2017

zhuravljov added a commit that referenced this pull request Dec 24, 2017

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment
You can’t perform that action at this time.