Merge branch 'master' into mongodb_transport

Author: makasim

.travis.yml

@@ -7,16 +7,12 @@ language: php
 
 matrix:
   include:
-#    - php: 5.6
-#      env: SYMFONY_VERSION=2.8.* UNIT_TESTS=true
     - php: 7.1
       env: SYMFONY_VERSION=3.0.* PHPSTAN=true
     - php: 7.1
       env: SYMFONY_VERSION=3.0.* PHP_CS_FIXER=true
     - php: 7.0
       env: SYMFONY_VERSION=2.8.* UNIT_TESTS=true
-    - php: 5.6
-      env: SYMFONY_VERSION=3.0.* UNIT_TESTS=true
     - php: 7.0
       env: SYMFONY_VERSION=3.0.* UNIT_TESTS=true
     - php: 7.1

CHANGELOG.md

@@ -1,5 +1,13 @@
 # Change Log
 
+## [0.8.27](https://github.com/php-enqueue/enqueue-dev/tree/0.8.27) (2018-05-01)
+[Full Changelog](https://github.com/php-enqueue/enqueue-dev/compare/0.8.26...0.8.27)
+
+- Kafka symfony transport [\#432](https://github.com/php-enqueue/enqueue-dev/pull/432) ([dheineman](https://github.com/dheineman))
+- Drop PHP5 support, Drop Symfony 2.X support.  [\#419](https://github.com/php-enqueue/enqueue-dev/pull/419) ([makasim](https://github.com/makasim))
+
+- How can I use the Symfony Bundle with Kafka? [\#428](https://github.com/php-enqueue/enqueue-dev/issues/428)
+
 ## [0.8.26](https://github.com/php-enqueue/enqueue-dev/tree/0.8.26) (2018-04-19)
 [Full Changelog](https://github.com/php-enqueue/enqueue-dev/compare/0.8.25...0.8.26)
 

docs/bundle/config_reference.md

@@ -21,8 +21,6 @@ enqueue:
             connection_timeout:   1
             buffer_size:          1000
             lazy:                 true
-            
-            # Should be true if you want to use secure connections. False by default
             ssl_on:               false
         rabbitmq_stomp:
             host:                 localhost
@@ -34,6 +32,7 @@ enqueue:
             connection_timeout:   1
             buffer_size:          1000
             lazy:                 true
+            ssl_on:               false
 
             # The option tells whether RabbitMQ broker has management plugin installed or not
             management_plugin_installed: false
@@ -42,7 +41,7 @@ enqueue:
             # The option tells whether RabbitMQ broker has delay plugin installed or not
             delay_plugin_installed: false
         amqp:
-            driver:               ~ # One of "ext"; "lib"; "bunny"
+            driver:               ~
 
             # The connection to AMQP broker set as a string. Other parameters could be used as defaults
             dsn:                  ~
@@ -106,7 +105,7 @@ enqueue:
             # Path to local private key file on filesystem in case of separate files for certificate (local_cert) and private key. A string.
             ssl_key:              ~
         rabbitmq_amqp:
-            driver:               ~ # One of "ext"; "lib"; "bunny"
+            driver:               ~
 
             # The connection to AMQP broker set as a string. Other parameters could be used as defaults
             dsn:                  ~
@@ -190,18 +189,27 @@ enqueue:
             polling_interval:     100
         redis:
 
+            # The redis connection given as DSN. For example redis://host:port?vendor=predis
+            dsn:                  ~
+
             # can be a host, or the path to a unix domain socket
-            host:                 ~ # Required
+            host:                 ~
             port:                 ~
 
             # The library used internally to interact with Redis server
-            vendor:               ~ # One of "phpredis"; "predis", Required
+            vendor:               ~ # One of "phpredis"; "predis"; "custom"
+
+            # A custom redis service id, used with vendor true only
+            redis:                ~
 
             # bool, Whether it use single persisted connection or open a new one for every context
             persisted:            false
 
             # the connection will be performed as later as possible, if the option set to true
             lazy:                 true
+
+            # Database index to select when connected.
+            database:             0
         dbal:
 
             # The Doctrine DBAL DSN. Other parameters are ignored if set
@@ -229,6 +237,7 @@ enqueue:
 
             # the connection will be performed as later as possible, if the option set to true
             lazy:                 true
+            endpoint:             null
         gps:
 
             # The connection to Google Pub/Sub broker set as a string. Other parameters are ignored if set
@@ -248,8 +257,36 @@ enqueue:
 
             # The connection will be performed as later as possible, if the option set to true
             lazy:                 true
+        rdkafka:
+
+            # The kafka DSN. Other parameters are ignored if set
+            dsn:                  ~
+
+            # The kafka global configuration properties
+            global:               []
+
+            # The kafka topic configuration properties
+            topic:                []
+
+            # Delivery report callback
+            dr_msg_cb:            ~
+
+            # Error callback
+            error_cb:             ~
+
+            # Called after consumer group has been rebalanced
+            rebalance_cb:         ~
+
+            # Which partitioner to use
+            partitioner:          ~ # One of "RD_KAFKA_MSG_PARTITIONER_RANDOM"; "RD_KAFKA_MSG_PARTITIONER_CONSISTENT"
+
+            # Logging level (syslog(3) levels)
+            log_level:            ~
+
+            # Commit asynchronous
+            commit_async:         false
     client:
-        traceable_producer:   false
+        traceable_producer:   true
         prefix:               enqueue
         app_name:             app
         router_topic:         default

docs/client/supported_brokers.md

@@ -13,23 +13,24 @@ Here's the list of transports supported by Enqueue Client:
 | Redis               | [enqueue/gps](../transport/redis.md)                       | redis:                          |
 | Amazon SQS          | [enqueue/sqs](../transport/sqs.md)                         | sqs:                            |
 | STOMP, RabbitMQ     | [enqueue/stomp](../transport/stomp.md)                     | stomp:                          |
+| Kafka               | [enqueue/stomp](../transport/kafka.md)                     | kafka:                          |
 | Null                | [enqueue/null](../transport/null.md)                       | null:                           |
 
 Here's the list of protocols and Client features supported by them 
 
 | Protocol       | Priority | Delay    | Expiration | Setup broker | Message bus |
 |:--------------:|:--------:|:--------:|:----------:|:------------:|:-----------:|
 | AMQP           |   No     |    No    |    Yes     |     Yes      |     Yes     |        
-| RabbitMQ AMQP  |   Yes    |    Yes*  |    Yes     |     Yes      |     Yes     |
+| RabbitMQ AMQP  |   Yes    |    Yes   |    Yes     |     Yes      |     Yes     |
 | STOMP          |   No     |    No    |    Yes     |     No       |     Yes**   |
-| RabbitMQ STOMP |   Yes    |    Yes*  |    Yes     |     Yes***   |     Yes**   |
+| RabbitMQ STOMP |   Yes    |    Yes   |    Yes     |     Yes***   |     Yes**   |
 | Filesystem     |   No     |    No    |    No      |     Yes      |     No      |
 | Redis          |   No     |    No    |    No      |  Not needed  |     No      |
 | Doctrine DBAL  |   Yes    |    Yes   |    No      |     Yes      |     No      |
 | Amazon SQS     |   No     |    Yes   |    No      |     Yes      |   Not impl  |
+| Kafka          |   No     |    No    |    No      |     Yes      |     No      |
 | Google PubSub  | Not impl | Not impl |  Not impl  |     Yes      |   Not impl  |
 
-* \* Possible if a RabbitMQ delay plugin is installed.
 * \*\* Possible if topics (exchanges) are configured on broker side manually.
 * \*\*\* Possible if RabbitMQ Management Plugin is installed.
 

pkg/dbal/DbalConsumer.php

@@ -4,7 +4,6 @@
 
 use Doctrine\DBAL\Connection;
 use Doctrine\DBAL\Types\Type;
-use Enqueue\Util\JSON;
 use Interop\Queue\InvalidMessageException;
 use Interop\Queue\PsrConsumer;
 use Interop\Queue\PsrMessage;

pkg/dbal/DbalProducer.php

@@ -3,7 +3,6 @@
 namespace Enqueue\Dbal;
 
 use Doctrine\DBAL\Types\Type;
-use Enqueue\Util\JSON;
 use Interop\Queue\Exception;
 use Interop\Queue\InvalidDestinationException;
 use Interop\Queue\InvalidMessageException;

pkg/dbal/JSON.php

@@ -0,0 +1,59 @@
+<?php
+
+namespace Enqueue\Dbal;
+
+class JSON
+{
+    /**
+     * @param string $string
+     *
+     * @throws \InvalidArgumentException
+     *
+     * @return array
+     */
+    public static function decode($string)
+    {
+        if (!is_string($string)) {
+            throw new \InvalidArgumentException(sprintf(
+                'Accept only string argument but got: "%s"',
+                is_object($string) ? get_class($string) : gettype($string)
+            ));
+        }
+
+        // PHP7 fix - empty string and null cause syntax error
+        if (empty($string)) {
+            return null;
+        }
+
+        $decoded = json_decode($string, true);
+        if (JSON_ERROR_NONE !== json_last_error()) {
+            throw new \InvalidArgumentException(sprintf(
+                'The malformed json given. Error %s and message %s',
+                json_last_error(),
+                json_last_error_msg()
+            ));
+        }
+
+        return $decoded;
+    }
+
+    /**
+     * @param mixed $value
+     *
+     * @return string
+     */
+    public static function encode($value)
+    {
+        $encoded = json_encode($value, JSON_UNESCAPED_UNICODE);
+
+        if (JSON_ERROR_NONE !== json_last_error()) {
+            throw new \InvalidArgumentException(sprintf(
+                'Could not encode value into json. Error %s and message %s',
+                json_last_error(),
+                json_last_error_msg()
+            ));
+        }
+
+        return $encoded;
+    }
+}

pkg/enqueue-bundle/EnqueueBundle.php

@@ -21,6 +21,8 @@
 use Enqueue\Fs\Symfony\FsTransportFactory;
 use Enqueue\Gps\GpsConnectionFactory;
 use Enqueue\Gps\Symfony\GpsTransportFactory;
+use Enqueue\RdKafka\RdKafkaConnectionFactory;
+use Enqueue\RdKafka\Symfony\RdKafkaTransportFactory;
 use Enqueue\Redis\RedisConnectionFactory;
 use Enqueue\Redis\Symfony\RedisTransportFactory;
 use Enqueue\Sqs\SqsConnectionFactory;
@@ -104,6 +106,12 @@ class_exists(AmqpLibConnectionFactory::class)
             $extension->setTransportFactory(new MissingTransportFactory('gps', ['enqueue/gps']));
         }
 
+        if (class_exists(RdKafkaConnectionFactory::class)) {
+            $extension->setTransportFactory(new RdKafkaTransportFactory('rdkafka'));
+        } else {
+            $extension->setTransportFactory(new MissingTransportFactory('rdkafka', ['enqueue/rdkafka']));
+        }
+
         $container->addCompilerPass(new AsyncEventsPass(), PassConfig::TYPE_BEFORE_OPTIMIZATION, 100);
         $container->addCompilerPass(new AsyncTransformersPass(), PassConfig::TYPE_BEFORE_OPTIMIZATION, 100);
     }

pkg/enqueue/Symfony/DefaultTransportFactory.php

@@ -10,6 +10,8 @@
 use Enqueue\Gps\Symfony\GpsTransportFactory;
 use Enqueue\Null\NullConnectionFactory;
 use Enqueue\Null\Symfony\NullTransportFactory;
+use Enqueue\RdKafka\RdKafkaConnectionFactory;
+use Enqueue\RdKafka\Symfony\RdKafkaTransportFactory;
 use Enqueue\Redis\RedisConnectionFactory;
 use Enqueue\Redis\Symfony\RedisTransportFactory;
 use Enqueue\Sqs\SqsConnectionFactory;
@@ -209,6 +211,10 @@ private function findFactory($dsn)
             return new StompTransportFactory('default_stomp');
         }
 
+        if ($factory instanceof RdKafkaConnectionFactory) {
+            return new RdKafkaTransportFactory('default_kafka');
+        }
+
         throw new \LogicException(sprintf(
             'There is no supported transport factory for the connection factory "%s" created from DSN "%s"',
             get_class($factory),

pkg/enqueue/Tests/Symfony/DefaultTransportFactoryTest.php

@@ -289,5 +289,7 @@ public static function provideDSNs()
         yield ['redis:', 'default_redis'];
 
         yield ['stomp:', 'default_stomp'];
+
+        yield ['kafka:', 'default_kafka'];
     }
 }