Merge remote-tracking branch 'origin/master' into pr-452

Author: makasim

CHANGELOG.md

@@ -1,5 +1,38 @@
 # Change Log
 
+## [0.8.33](https://github.com/php-enqueue/enqueue-dev/tree/0.8.33) (2018-07-26)
+[Full Changelog](https://github.com/php-enqueue/enqueue-dev/compare/0.8.32...0.8.33)
+
+- \[consumption\] process niceness extension [\#449](https://github.com/php-enqueue/enqueue-dev/issues/449)
+- \[Symfony\] AsyncListener does not use TraceableProducer [\#392](https://github.com/php-enqueue/enqueue-dev/issues/392)
+
+- Fix call debug method on null [\#480](https://github.com/php-enqueue/enqueue-dev/pull/480) ([makasim](https://github.com/makasim))
+
+- Support MQTT [\#477](https://github.com/php-enqueue/enqueue-dev/issues/477)
+- Bugs in RabbitMqDelayPluginDelayStrategy [\#455](https://github.com/php-enqueue/enqueue-dev/issues/455)
+- \[sqs\] Support using a pre-configured SqsClient [\#443](https://github.com/php-enqueue/enqueue-dev/issues/443)
+- IronMQ \(iron.io\) provider ? [\#415](https://github.com/php-enqueue/enqueue-dev/issues/415)
+
+- Add Localstack Docker container for SQS functional tests [\#473](https://github.com/php-enqueue/enqueue-dev/pull/473) ([elazar](https://github.com/elazar))
+- \[consumption\] add process niceness extension [\#467](https://github.com/php-enqueue/enqueue-dev/pull/467) ([ramunasd](https://github.com/ramunasd))
+
+## [0.8.32](https://github.com/php-enqueue/enqueue-dev/tree/0.8.32) (2018-07-10)
+[Full Changelog](https://github.com/php-enqueue/enqueue-dev/compare/0.8.31...0.8.32)
+
+- \[Bundle\] auto-tag services [\#409](https://github.com/php-enqueue/enqueue-dev/issues/409)
+
+- Add documentation the processor services need to be public [\#406](https://github.com/php-enqueue/enqueue-dev/issues/406)
+
+- Is it possible to read messages in batch? [\#472](https://github.com/php-enqueue/enqueue-dev/issues/472)
+- Batch publishing [\#463](https://github.com/php-enqueue/enqueue-dev/issues/463)
+- populating, missing messages and supervisor [\#460](https://github.com/php-enqueue/enqueue-dev/issues/460)
+- Processor was not found. processorName: "enqueue.client.router\_processor" [\#451](https://github.com/php-enqueue/enqueue-dev/issues/451)
+- \[Bundle\] Enqueue\Symfony\Client\ContainerAwareProcessorRegistry expects processors to be public [\#410](https://github.com/php-enqueue/enqueue-dev/issues/410)
+
+- Update of "back to index" link [\#468](https://github.com/php-enqueue/enqueue-dev/pull/468) ([N-M](https://github.com/N-M))
+- PHP\_URL\_SCHEME doesn't support underscores [\#453](https://github.com/php-enqueue/enqueue-dev/pull/453) ([coudenysj](https://github.com/coudenysj))
+- WIP: Add support for using a pre-configured client with the SQS driver [\#444](https://github.com/php-enqueue/enqueue-dev/pull/444) ([elazar](https://github.com/elazar))
+
 ## [0.8.31](https://github.com/php-enqueue/enqueue-dev/tree/0.8.31) (2018-05-24)
 [Full Changelog](https://github.com/php-enqueue/enqueue-dev/compare/0.8.30...0.8.31)
 

bin/run-fun-test.sh

@@ -3,4 +3,4 @@
 set -x
 set -e
 
-docker-compose run --workdir="/mqdev" --rm dev ./bin/test "$@"
+docker-compose run --workdir="/mqdev" --rm dev ./bin/test "$@"

bin/test

@@ -37,6 +37,7 @@ waitForService beanstalkd 11300 50
 waitForService gearmand 4730 50
 waitForService kafka 9092 50
 waitForService mongo 27017 50
+waitForService localstack 4576 50
 
 php pkg/job-queue/Tests/Functional/app/console doctrine:database:create --if-not-exists
 php pkg/job-queue/Tests/Functional/app/console doctrine:schema:update --force

docker-compose.yml

@@ -14,6 +14,7 @@ services:
       - google-pubsub
       - rabbitmqssl
       - mongo
+      - localstack
     volumes:
       - './:/mqdev'
     environment:
@@ -34,9 +35,11 @@ services:
       - DOCTRINE_PASSWORD=rootpass
       - REDIS_HOST=redis
       - REDIS_PORT=6379
-      - AWS_SQS_KEY=$ENQUEUE_AWS__SQS__KEY
-      - AWS_SQS_SECRET=$ENQUEUE_AWS__SQS__SECRET
-      - AWS_SQS_REGION=$ENQUEUE_AWS__SQS__REGION
+      - AWS_SQS_KEY=key
+      - AWS_SQS_SECRET=secret
+      - AWS_SQS_REGION=us-east-1
+      - AWS_SQS_ENDPOINT=http://localstack:4576
+      - AWS_SQS_VERSION=latest
       - BEANSTALKD_HOST=beanstalkd
       - BEANSTALKD_PORT=11300
       - BEANSTALKD_DSN=beanstalk://beanstalkd:11300
@@ -109,6 +112,14 @@ services:
       ports:
         - "27017:27017"
 
+  localstack:
+    image: 'localstack/localstack:latest'
+    ports:
+      - '4576:4576'
+    environment:
+      HOSTNAME_EXTERNAL: 'localstack'
+      SERVICES: 'sqs'
+
 volumes:
   mysql-data:
     driver: local

docs/bundle/config_reference.md

@@ -228,6 +228,7 @@ enqueue:
             polling_interval:     1000
             lazy:                 true
         sqs:
+            client:               null
             key:                  null
             secret:               null
             token:                null

docs/contribution.md

@@ -1,8 +1,8 @@
 # Contribution
 
-To contribute you have to fork a [enqueue-dev](https://github.com/php-enqueue/enqueue-dev) repository.
-Clone it locally.
- 
+To contribute you have to send a pull request to [enqueue-dev](https://github.com/php-enqueue/enqueue-dev) repository. 
+The pull requests to read only subtree split [repositories](https://github.com/php-enqueue/enqueue-dev/blob/master/bin/subtree-split#L46) will be closed. 
+
 ## Setup environment
 
 ```
@@ -15,15 +15,22 @@ Once you did it you can work on a feature or bug fix.
 
 ## Testing
 
-To run tests simply run 
+To run tests
 
 ```
 ./bin/dev -t
 ```
 
+or for a package only:
+
+
+```
+./bin/dev -t pkg/enqueue
+```
+
 ## Commit 
 
 When you try to commit changes `php-cs-fixer` is run. It fixes all coding style issues. Don't forget to stage them and commit everything.
 Once everything is done open a pull request on official repository. 
 
-[back to index](index.md)
+[back to index](index.md)

docs/cookbook/symfony/how-to-change-consume-command-logger.md

@@ -22,7 +22,7 @@ services:
 
 The logger extension with the highest priority will set its logger.
 
-[back to index](../index.md)  
+[back to index](../../index.md)  
 
 
 

docs/index.md

@@ -75,3 +75,4 @@
 * [Symfony Async EventDispatcher](https://blog.forma-pro.com/symfony-async-eventdispatcher-d01055a255cf)
 * [Spool Swiftmailer emails to real message queue.](https://blog.forma-pro.com/spool-swiftmailer-emails-to-real-message-queue-9ecb8b53b5de)
 * [Yii PHP Framework has adopted AMQP Interop.](https://blog.forma-pro.com/yii-php-framework-has-adopted-amqp-interop-85ab47c9869f)
+* [(En)queue Symfony console commands](http://tech.yappa.be/enqueue-symfony-console-commands)

docs/transport/sqs.md

@@ -29,11 +29,15 @@ $factory = new SqsConnectionFactory([
     'region' => 'aRegion',
 ]);
 
-// same as above but given as DSN string
+// same as above but given as DSN string. You may need to url encode secret if it contains special char (like +)
 $factory = new SqsConnectionFactory('sqs:?key=aKey&secret=aSecret&region=aRegion');
 
 $psrContext = $factory->createContext();
 
+// using a pre-configured client
+$client = new Aws\Sqs\SqsClient([ /* ... */ ]);
+$factory = new SqsConnectionFactory($client);
+
 // if you have enqueue/enqueue library installed you can use a function from there to create the context
 $psrContext = \Enqueue\dsn_to_context('sqs:');
 ```
@@ -109,4 +113,4 @@ $fooQueue = $psrContext->createQueue('foo');
 $psrContext->purge($fooQueue);
 ```
 
-[back to index](../index.md)
+[back to index](../index.md)

pkg/amqp-ext/AmqpContext.php

@@ -315,7 +315,7 @@ public function subscribe(InteropAmqpConsumer $consumer, callable $callback)
 
         $consumerTag = $extQueue->getConsumerTag();
         $consumer->setConsumerTag($consumerTag);
-        $this->subscribers[$consumerTag] = [$consumer, $callback];
+        $this->subscribers[$consumerTag] = [$consumer, $callback, $extQueue];
     }
 
     /**
@@ -327,15 +327,13 @@ public function unsubscribe(InteropAmqpConsumer $consumer)
             return;
         }
 
-        // seg fault
-//        $consumerTag = $consumer->getConsumerTag();
-//        $consumer->setConsumerTag(null);
-//
-//        $extQueue = new \AMQPQueue($this->getExtChannel());
-//        $extQueue->setName($consumer->getQueue()->getQueueName());
-//
-//        $extQueue->cancel($consumerTag);
-//        unset($this->subscribers[$consumerTag]);
+        $consumerTag = $consumer->getConsumerTag();
+        $consumer->setConsumerTag(null);
+
+        list($consumer, $callback, $extQueue) = $this->subscribers[$consumerTag];
+
+        $extQueue->cancel($consumerTag);
+        unset($this->subscribers[$consumerTag]);
     }
 
     /**

pkg/dbal/DbalConnectionFactory.php

@@ -98,10 +98,10 @@ private function parseDsn($dsn)
             throw new \LogicException(sprintf('Failed to parse DSN "%s"', $dsn));
         }
 
-        $schema = parse_url($dsn, PHP_URL_SCHEME);
-        if (empty($schema)) {
+        if (!preg_match('/^([0-9a-z_]+):(.+)?$/', $dsn, $matches)) {
             throw new \LogicException('Schema is empty');
         }
+        $schema = $matches[1];
 
         $supported = [
             'db2' => true,

pkg/dbal/Tests/DbalConnectionFactoryConfigTest.php

@@ -72,6 +72,16 @@ public static function provideConfigs()
             ],
         ];
 
+        yield [
+            'pdo_mysql:',
+            [
+                'lazy' => true,
+                'connection' => [
+                    'url' => 'pdo_mysql://root@localhost',
+                ],
+            ],
+        ];
+
         yield [
             'pgsql:',
             [
@@ -92,6 +102,16 @@ public static function provideConfigs()
             ],
         ];
 
+        yield [
+            'pdo_mysql://user:pass@host:10001/db',
+            [
+                'lazy' => true,
+                'connection' => [
+                    'url' => 'pdo_mysql://user:pass@host:10001/db',
+                ],
+            ],
+        ];
+
         yield [
             [],
             [

pkg/enqueue-bundle/Tests/Functional/App/config/custom-config.yml

@@ -1,7 +1,11 @@
 parameters:
     locale: 'en'
     secret: 'ThisTokenIsNotSoSecretChangeIt'
-
+    env(AWS_SQS_REGION): 'us-east-1'
+    env(AWS_SQS_VERSION): 'latest'
+    env(AWS_SQS_KEY): 'key'
+    env(AWS_SQS_SECRET): 'secret'
+    env(AWS_SQS_ENDPOINT): 'http://localstack:4576'
 
 framework:
     #esi:             ~
@@ -33,3 +37,15 @@ services:
         public: true
         tags:
             - { name: 'enqueue.client.processor' }
+
+    test.sqs_client:
+        public: true
+        class: Aws\Sqs\SqsClient
+        arguments:
+            -
+                endpoint: '%env(AWS_SQS_ENDPOINT)%'
+                region: '%env(AWS_SQS_REGION)%'
+                version: '%env(AWS_SQS_VERSION)%'
+                credentials:
+                    key: '%env(AWS_SQS_KEY)%'
+                    secret: '%env(AWS_SQS_SECRET)%'

pkg/enqueue-bundle/Tests/Functional/UseCasesTest.php

@@ -201,6 +201,16 @@ public function provideEnqueueConfigs()
                         'key' => getenv('AWS_SQS_KEY'),
                         'secret' => getenv('AWS_SQS_SECRET'),
                         'region' => getenv('AWS_SQS_REGION'),
+                        'endpoint' => getenv('AWS_SQS_ENDPOINT'),
+                    ],
+                ],
+            ]];
+
+            yield 'sqs_client' => [[
+                'transport' => [
+                    'default' => 'sqs',
+                    'sqs' => [
+                        'client' => 'test.sqs_client',
                     ],
                 ],
             ]];

pkg/enqueue/Consumption/Extension/NicenessExtension.php

@@ -0,0 +1,50 @@
+<?php
+
+namespace Enqueue\Consumption\Extension;
+
+use Enqueue\Consumption\Context;
+use Enqueue\Consumption\EmptyExtensionTrait;
+use Enqueue\Consumption\ExtensionInterface;
+
+class NicenessExtension implements ExtensionInterface
+{
+    use EmptyExtensionTrait;
+
+    /**
+     * @var int
+     */
+    protected $niceness = 0;
+
+    /**
+     * @param int $niceness
+     *
+     * @throws \InvalidArgumentException
+     */
+    public function __construct($niceness)
+    {
+        if (false === is_int($niceness)) {
+            throw new \InvalidArgumentException(sprintf(
+                'Expected niceness value is int but got: "%s"',
+                is_object($niceness) ? get_class($niceness) : gettype($niceness)
+            ));
+        }
+
+        $this->niceness = $niceness;
+    }
+
+    /**
+     * {@inheritdoc}
+     */
+    public function onStart(Context $context)
+    {
+        if (0 !== $this->niceness) {
+            $changed = @proc_nice($this->niceness);
+            if (!$changed) {
+                throw new \InvalidArgumentException(sprintf(
+                    'Cannot change process niceness, got warning: "%s"',
+                    error_get_last()['message']
+                ));
+            }
+        }
+    }
+}

pkg/enqueue/Consumption/QueueConsumer.php

@@ -178,7 +178,13 @@ public function consume(ExtensionInterface $runtimeExtension = null)
         $context = new Context($this->psrContext);
         $this->extension->onStart($context);
 
-        $this->logger = $context->getLogger() ?: new NullLogger();
+        if ($context->getLogger()) {
+            $this->logger = $context->getLogger();
+        } else {
+            $this->logger = new NullLogger();
+            $context->setLogger($this->logger);
+        }
+
         $this->logger->info('Start consuming');
 
         if ($this->psrContext instanceof AmqpContext) {

pkg/enqueue/Symfony/Consumption/LimitsExtensionsCommandTrait.php

@@ -5,6 +5,7 @@
 use Enqueue\Consumption\Extension\LimitConsumedMessagesExtension;
 use Enqueue\Consumption\Extension\LimitConsumerMemoryExtension;
 use Enqueue\Consumption\Extension\LimitConsumptionTimeExtension;
+use Enqueue\Consumption\Extension\NicenessExtension;
 use Enqueue\Consumption\ExtensionInterface;
 use Symfony\Component\Console\Input\InputInterface;
 use Symfony\Component\Console\Input\InputOption;
@@ -20,7 +21,8 @@ protected function configureLimitsExtensions()
         $this
             ->addOption('message-limit', null, InputOption::VALUE_REQUIRED, 'Consume n messages and exit')
             ->addOption('time-limit', null, InputOption::VALUE_REQUIRED, 'Consume messages during this time')
-            ->addOption('memory-limit', null, InputOption::VALUE_REQUIRED, 'Consume messages until process reaches this memory limit in MB');
+            ->addOption('memory-limit', null, InputOption::VALUE_REQUIRED, 'Consume messages until process reaches this memory limit in MB')
+            ->addOption('niceness', null, InputOption::VALUE_REQUIRED, 'Set process niceness');
     }
 
     /**
@@ -58,6 +60,11 @@ protected function getLimitsExtensions(InputInterface $input, OutputInterface $o
             $extensions[] = new LimitConsumerMemoryExtension($memoryLimit);
         }
 
+        $niceness = $input->getOption('niceness');
+        if ($niceness) {
+            $extensions[] = new NicenessExtension($niceness);
+        }
+
         return $extensions;
     }
 }