Laravel 6 support

.travis.yml

@@ -1,11 +1,18 @@
 language: php
 
 php:
-  - 5.6
-  - 7.0
-  - 7.1
+  - 7.2
+  - 7.3
+  - 7.4
+  - hhvm
   - nightly
 
+matrix:
+    allow_failures:
+        - php: 7.4
+        - php: hhvm
+        - php: nightly
+
 before_script:
   - composer install
 

README.md

@@ -1,13 +1,15 @@
 # laravel-pubsub
 
 A Pub-Sub abstraction for Laravel.
+_forked from Superbalist/laravel-pubsub_
+
 
 [![Author](http://img.shields.io/badge/[email protected]?style=flat-square)](https://twitter.com/superbalist)
-[![Build Status](https://img.shields.io/travis/Superbalist/laravel-pubsub/master.svg?style=flat-square)](https://travis-ci.org/Superbalist/laravel-pubsub)
+[![Build Status](https://img.shields.io/travis/Entanet/laravel-pubsub/master.svg?style=flat-square)](https://travis-ci.org/Entanet/laravel-pubsub)
 [![StyleCI](https://styleci.io/repos/67405993/shield?branch=master)](https://styleci.io/repos/67405993)
 [![Software License](https://img.shields.io/badge/license-MIT-brightgreen.svg?style=flat-square)](LICENSE)
-[![Packagist Version](https://img.shields.io/packagist/v/superbalist/laravel-pubsub.svg?style=flat-square)](https://packagist.org/packages/superbalist/laravel-pubsub)
-[![Total Downloads](https://img.shields.io/packagist/dt/superbalist/laravel-pubsub.svg?style=flat-square)](https://packagist.org/packages/superbalist/laravel-pubsub)
+[![Packagist Version](https://img.shields.io/packagist/v/entanet/laravel-pubsub.svg?style=flat-square)](https://packagist.org/packages/entanet/laravel-pubsub)
+[![Total Downloads](https://img.shields.io/packagist/dt/entanet/laravel-pubsub.svg?style=flat-square)](https://packagist.org/packages/entanet/laravel-pubsub)
 
 This package is a wrapper bridging [php-pubsub](https://github.com/Superbalist/php-pubsub) into Laravel.
 
@@ -28,7 +30,7 @@ The following adapters are supported:
 ## Installation
 
 ```bash
-composer require superbalist/laravel-pubsub
+composer require entanet/laravel-pubsub
 ```
 
 Register the service provider in app.php
@@ -56,6 +58,9 @@ REDIS_PASSWORD=null
 REDIS_PORT=6379
 
 KAFKA_BROKERS=localhost
+KAFKA_SECURITY_PROTOCOL=kafka-security-protocol
+KAFKA_SASL_USERNAME=kafka-sasl-username
+KAFKA_SASL_PASSWORD=kafks-sasl-password
 
 GOOGLE_CLOUD_PROJECT_ID=your-project-id-here
 GOOGLE_CLOUD_KEY_FILE=path/to/your/gcloud-key.json
@@ -64,6 +69,8 @@ HTTP_PUBSUB_URI=null
 HTTP_PUBSUB_SUBSCRIBE_CONNECTION=redis
 ```
 
+If the KAFKA_SECURITY_PROTOCOL is set to either "SASL_SSL" or "SASL_PLAINTEXT", the credentials stored in KAFKA_SASL_USERNAME & KAFKA_SASL_PASSWORD will be used to authenticate with the kafka server / cluster.
+
 To customize the configuration file, publish the package configuration using Artisan.
 ```bash
 php artisan vendor:publish --provider="Superbalist\LaravelPubSub\PubSubServiceProvider"

changelog.md

@@ -1,5 +1,16 @@
 # Changelog
 
+## 3.2.1 - 2019-10-07
+* Bugfix, remove call to setDetaulTopicConf() in RdKafka\Conf, which was deprecated in rdkafka 4.0.0 (pecl)
+
+## 3.2.0 - 2019-09-24
+
+* Allow enabling SASL by setting KAFKA_SECURITY_PROTOCOL env variable
+
+## 3.1.0 - 2019-07-23
+
+* Add support for SASL authentication
+
 ## 3.0.0 - 2017-07-25
 
 * Bump up to superbalist/php-pubsub-google-cloud ^5.0 which allows for background daemon support

composer.json

@@ -1,5 +1,5 @@
 {
-  "name": "superbalist/laravel-pubsub",
+  "name": "entanet/laravel-pubsub",
   "description": "A Pub-Sub abstraction for Laravel",
   "license": "MIT",
   "authors": [
@@ -8,13 +8,16 @@
       "email": "[email protected]"
     }
   ],
+  "conflict": {
+    "superbalist/laravel-pubsub": "*"
+  },
   "require": {
-    "php": ">=5.6.0",
-    "illuminate/support": "^5.4",
+    "php": ">=7.2.0",
+    "illuminate/support": "^6",
     "superbalist/php-pubsub": "^1.0|^2.0",
     "superbalist/php-pubsub-redis": "^1.0|^2.0",
     "superbalist/php-pubsub-google-cloud": "^5.0",
-    "illuminate/console": "^5.4",
+    "illuminate/console": "^6",
     "superbalist/php-pubsub-http": "^1.0"
   },
   "autoload": {

config/pubsub.php

@@ -48,8 +48,11 @@
 
         'kafka' => [
             'driver' => 'kafka',
-            'consumer_group_id' => 'php-pubsub',
+            'consumer_group_id' => env('KAFKA_CONSUMER_ID', 'php-pubsub'),
             'brokers' => env('KAFKA_BROKERS', 'localhost'),
+            'security_protocol' => env('KAFKA_SECURITY_PROTOCOL'),
+            'sasl_username' => env('KAFKA_SASL_USERNAME'),
+            'sasl_password' => env('KAFKA_SASL_PASSWORD'),
         ],
 
         'gcloud' => [

src/PubSubConnectionFactory.php

@@ -3,6 +3,7 @@
 namespace Superbalist\LaravelPubSub;
 
 use Illuminate\Contracts\Container\Container;
+use Illuminate\Support\Arr;
 use InvalidArgumentException;
 use Superbalist\PubSub\Adapters\DevNullPubSubAdapter;
 use Superbalist\PubSub\Adapters\LocalPubSubAdapter;
@@ -82,21 +83,39 @@ protected function makeRedisAdapter(array $config)
      */
     protected function makeKafkaAdapter(array $config)
     {
-        // create producer
-        $producer = $this->container->makeWith('pubsub.kafka.producer');
-        $producer->addBrokers($config['brokers']);
-
-        // create consumer
+        // create default topic
         $topicConf = $this->container->makeWith('pubsub.kafka.topic_conf');
         $topicConf->set('auto.offset.reset', 'smallest');
 
+        // create config
         $conf = $this->container->makeWith('pubsub.kafka.conf');
-        $conf->set('group.id', array_get($config, 'consumer_group_id', 'php-pubsub'));
+        $conf->set('group.id', Arr::get($config, 'consumer_group_id', 'php-pubsub'));
         $conf->set('metadata.broker.list', $config['brokers']);
         $conf->set('enable.auto.commit', 'false');
         $conf->set('offset.store.method', 'broker');
-        $conf->setDefaultTopicConf($topicConf);
 
+        if (array_key_exists('security_protocol', $config)
+            && $config['security_protocol'] != ''
+        ) {
+            switch ($config['security_protocol']) {
+                case 'SASL_SSL':
+                case 'SASL_PLAINTEXT':
+                    $conf->set('security.protocol', Arr::get($config, 'security_protocol', 'SASL_SSL'));
+                    $conf->set('sasl.username', Arr::get($config, 'sasl_username', 'sasl_username'));
+                    $conf->set('sasl.password', Arr::get($config, 'sasl_password', 'sasl_password'));
+                    $conf->set('sasl.mechanisms', Arr::get($config, 'sasl_mechanisms', 'PLAIN'));
+                    break;
+
+                default:
+                    break;
+            }
+        }
+
+        // create producer
+        $producer = $this->container->makeWith('pubsub.kafka.producer', ['conf' => $conf]);
+        $producer->addBrokers($config['brokers']);
+
+        // create consumer
         $consumer = $this->container->makeWith('pubsub.kafka.consumer', ['conf' => $conf]);
 
         return new KafkaPubSubAdapter($producer, $consumer);
@@ -121,11 +140,11 @@ protected function makeGoogleCloudAdapter(array $config)
 
         $client = $this->container->makeWith('pubsub.gcloud.pub_sub_client', ['config' => $clientConfig]);
 
-        $clientIdentifier = array_get($config, 'client_identifier');
-        $autoCreateTopics = array_get($config, 'auto_create_topics', true);
-        $autoCreateSubscriptions = array_get($config, 'auto_create_subscriptions', true);
-        $backgroundBatching = array_get($config, 'background_batching', false);
-        $backgroundDaemon = array_get($config, 'background_daemon', false);
+        $clientIdentifier = Arr::get($config, 'client_identifier');
+        $autoCreateTopics = Arr::get($config, 'auto_create_topics', true);
+        $autoCreateSubscriptions = Arr::get($config, 'auto_create_subscriptions', true);
+        $backgroundBatching = Arr::get($config, 'background_batching', false);
+        $backgroundDaemon = Arr::get($config, 'background_daemon', false);
 
         if ($backgroundDaemon) {
             putenv('IS_BATCH_DAEMON_RUNNING=true');

src/PubSubManager.php

@@ -3,6 +3,7 @@
 namespace Superbalist\LaravelPubSub;
 
 use Illuminate\Contracts\Foundation\Application;
+use Illuminate\Support\Arr;
 use InvalidArgumentException;
 use Superbalist\PubSub\PubSubAdapterInterface;
 
@@ -79,7 +80,7 @@ protected function makeConnection($name)
             );
         }
 
-        return $this->factory->make($config['driver'], array_except($config, ['driver']));
+        return $this->factory->make($config['driver'], Arr::except($config, ['driver']));
     }
 
     /**

src/PubSubServiceProvider.php

@@ -64,8 +64,8 @@ protected function registerAdapterDependencies()
             return new \RdKafka\TopicConf();
         });
 
-        $this->app->bind('pubsub.kafka.producer', function () {
-            return new \RdKafka\Producer();
+        $this->app->bind('pubsub.kafka.producer', function ($app, $parameters) {
+            return new \RdKafka\Producer($parameters['conf']);
         });
 
         $this->app->bind('pubsub.kafka.conf', function () {