Merge pull request #1 from SepteniTechnology/feature/add_ssl_configuration

feature/add_ssl_configuration

Author: nhatnv6

config/kafka.php

@@ -3,4 +3,14 @@
 return [
     // Brokers address
     'brokers' => array_filter(explode(',',env('KAFKA_BROKERS', ''))),
+    'options' => [
+        'log_level' => LOG_DEBUG,
+        'debug' => 'all',
+        'security.protocol' => 'ssl',
+        'ssl.truststore.type' => 'JKS',
+        'ssl.truststore.location' => env('KAFKA_SSL_TRUSTSTORE_LOCATION'),
+        'ssl.truststore.password' => env('KAFKA_SSL_TRUSTSTORE_PASSWORD'),
+        'ssl.keystore.location' => env('KAFKA_SSL_KEYSTORE_LOCATION'),
+        'ssl.keystore.password' => env('KAFKA_SSL_KEYSTORE_PASSWORD')
+    ]
 ];

src/KafkaServerProvider.php

@@ -21,10 +21,18 @@ public function register()
         });
 
 
-        $this->app->singleton(RdKafka::class, function () {
+        $this->app->singleton(RdKafka::class, function ($app) {
             $conf = new Conf();
-            $conf->set('log_level', (string) LOG_DEBUG);
-            $conf->set('debug', 'all');
+
+            $config = $app['config']->get('kafka');
+
+            $options = $config['options'] ?? [];
+            foreach ($options as $option => $value) {
+                $conf->set($option, $value);
+            }
+
+//            $conf->set('log_level', (string) LOG_DEBUG);
+//            $conf->set('debug', 'all');
             return new Producer($conf);
         });
 
@@ -34,4 +42,4 @@ public function register()
             __DIR__ . '/../config/kafka.php' => config_path('kafka.php')
         ], 'kafka');
     }
-}
+}