add brokers to config

@@ -240,6 +240,10 @@ kafka {
     # whether enabled kafka.
     # whether enabled kafka.
     # default: off
     # default: off
     enabled         off;
     enabled         off;
+    # the broker list, broker is <ip:port>
+    # and use space to specify multple brokers.
+    # for exampl, 127.0.0.1:9092 127.0.0.1:9093
+    brokers         127.0.0.1:9092;
 }
 }
 
 
 #############################################################################################
 #############################################################################################

@@ -2120,6 +2120,17 @@ int SrsConfig::global_to_json(SrsJsonObject* obj)
                 }
                 }
             }
             }
             obj->set(dir->name, sobj);
             obj->set(dir->name, sobj);
+        } else if (dir->name == "kafka") {
+            SrsJsonObject* sobj = SrsJsonAny::object();
+            for (int j = 0; j < (int)dir->directives.size(); j++) {
+                SrsConfDirective* sdir = dir->directives.at(j);
+                if (sdir->name == "enabled") {
+                    sobj->set(sdir->name, sdir->dumps_arg0_to_boolean());
+                } else if (sdir->name == "brokers") {
+                    sobj->set(sdir->name, sdir->dumps_args());
+                }
+            }
+            obj->set(dir->name, sobj);
         } else if (dir->name == "stream_caster") {
         } else if (dir->name == "stream_caster") {
             SrsJsonObject* sobj = SrsJsonAny::object();
             SrsJsonObject* sobj = SrsJsonAny::object();
             for (int j = 0; j < (int)dir->directives.size(); j++) {
             for (int j = 0; j < (int)dir->directives.size(); j++) {
@@ -3535,7 +3546,7 @@ int SrsConfig::check_config()
         SrsConfDirective* conf = root->get("kafka");
         SrsConfDirective* conf = root->get("kafka");
         for (int i = 0; conf && i < (int)conf->directives.size(); i++) {
         for (int i = 0; conf && i < (int)conf->directives.size(); i++) {
             string n = conf->at(i)->name;
             string n = conf->at(i)->name;
-            if (n != "enabled") {
+            if (n != "enabled" && n != "brokers") {
                 ret = ERROR_SYSTEM_CONFIG_INVALID;
                 ret = ERROR_SYSTEM_CONFIG_INVALID;
                 srs_error("unsupported kafka directive %s, ret=%d", n.c_str(), ret);
                 srs_error("unsupported kafka directive %s, ret=%d", n.c_str(), ret);
                 return ret;
                 return ret;
@@ -4272,6 +4283,21 @@ bool SrsConfig::get_kafka_enabled()
     return SRS_CONF_PERFER_FALSE(conf->arg0());
     return SRS_CONF_PERFER_FALSE(conf->arg0());
 }
 }
 
 
+SrsConfDirective* SrsConfig::get_kafka_brokers()
+{
+    SrsConfDirective* conf = root->get("kafka");
+    if (!conf) {
+        return NULL;
+    }
+    
+    conf->get("brokers");
+    if (!conf || conf->args.empty()) {
+        return NULL;
+    }
+    
+    return conf;
+}
+
 SrsConfDirective* SrsConfig::get_vhost(string vhost, bool try_default_vhost)
 SrsConfDirective* SrsConfig::get_vhost(string vhost, bool try_default_vhost)
 {
 {
     srs_assert(root);
     srs_assert(root);

@@ -634,6 +634,10 @@ public:
      * whether the kafka enabled.
      * whether the kafka enabled.
      */
      */
     virtual bool                get_kafka_enabled();
     virtual bool                get_kafka_enabled();
+    /**
+     * get the broker list, each is format in <ip:port>.
+     */
+    virtual SrsConfDirective*   get_kafka_brokers();
 // vhost specified section
 // vhost specified section
 public:
 public:
     /**
     /**

@@ -27,6 +27,7 @@
 #include <srs_kernel_log.hpp>
 #include <srs_kernel_log.hpp>
 #include <srs_app_config.hpp>
 #include <srs_app_config.hpp>
 #include <srs_app_async_call.hpp>
 #include <srs_app_async_call.hpp>
+#include <srs_app_utility.hpp>
 
 
 #ifdef SRS_AUTO_KAFKA
 #ifdef SRS_AUTO_KAFKA
 
 
@@ -64,7 +65,8 @@ int SrsKafkaProducer::start()
         return ret;
         return ret;
     }
     }
     
     
-    srs_trace("kafka worker ok, enabled:%d", _srs_config->get_kafka_enabled());
+    std::string enabled = srs_bool2switch(_srs_config->get_kafka_enabled());
+    srs_trace("kafka worker ok, enabled:%s", enabled.c_str());
     
     
     return ret;
     return ret;
 }
 }