Merge pull request #17 from sjje/master

add sasl authorization

Author: petermein

config/kafka.php

@@ -36,4 +36,24 @@
      * Sleep when a deadlock is detected
      */
     'sleep_on_deadlock' => env('KAFKA_DEADLOCK_SLEEP', 2),
+
+    /*
+     * sasl authorization
+     */
+    'sasl_enable' => false,
+
+    /*
+     * File or directory path to CA certificate(s) for verifying the broker's key. example: storage_path('kafka.client.truststore.jks')
+     */
+    'ssl_ca_location' => '',
+
+    /*
+     * SASL username for use with the PLAIN and SASL-SCRAM-.. mechanisms 
+     */
+    'sasl_plain_username' => env('KAFKA_SASL_PLAIN_USERNAME'),
+
+    /*
+     * SASL password for use with the PLAIN and SASL-SCRAM-.. mechanism 
+     */
+    'sasl_plain_password' => env('KAFKA_SASL_PLAIN_PASSWORD'),
 ];

src/Queue/Connectors/KafkaConnector.php

@@ -46,6 +46,12 @@ public function connect(array $config)
 
         /** @var Conf $conf */
         $conf = $this->container->makeWith('queue.kafka.conf', []);
+        if (true === $config['sasl_enable']) {
+            $conf->set('sasl.mechanisms', 'PLAIN');
+            $conf->set('sasl.username', $config['sasl_plain_username']);
+            $conf->set('sasl.password', $config['sasl_plain_password']);
+            $conf->set('ssl.ca.location', $config['ssl_ca_location']);
+        }
         $conf->set('group.id', array_get($config, 'consumer_group_id', 'php-pubsub'));
         $conf->set('metadata.broker.list', $config['brokers']);
         $conf->set('enable.auto.commit', 'false');