add load balance round robin for brokers.

trunk/src/app/srs_app_config.cpp

@@ -4290,7 +4290,7 @@ SrsConfDirective* SrsConfig::get_kafka_brokers()
         return NULL;
     }
     
-    conf->get("brokers");
+    conf = conf->get("brokers");
     if (!conf || conf->args.empty()) {
         return NULL;
     }

trunk/src/app/srs_app_kafka.cpp

@@ -23,21 +23,28 @@
 
 #include <srs_app_kafka.hpp>
 
+#include <vector>
+using namespace std;
+
 #include <srs_kernel_error.hpp>
 #include <srs_kernel_log.hpp>
 #include <srs_app_config.hpp>
 #include <srs_app_async_call.hpp>
 #include <srs_app_utility.hpp>
+#include <srs_kernel_utility.hpp>
+#include <srs_kernel_balance.hpp>
 
 #ifdef SRS_AUTO_KAFKA
 
 SrsKafkaProducer::SrsKafkaProducer()
 {
+    lb = new SrsLbRoundRobin();
     worker = new SrsAsyncCallWorker();
 }
 
 SrsKafkaProducer::~SrsKafkaProducer()
 {
+    srs_freep(lb);
     srs_freep(worker);
 }
 
@@ -46,7 +53,7 @@ int SrsKafkaProducer::initialize()
     int ret = ERROR_SUCCESS;
     
     // when kafka enabled, request metadata when startup.
-    if (_srs_config->get_kafka_enabled() && (ret = request_metadata()) != ERROR_SUCCESS) {
+    if ((ret = request_metadata()) != ERROR_SUCCESS) {
         srs_error("request kafka metadata failed. ret=%d", ret);
         return ret;
     }
@@ -65,8 +72,7 @@ int SrsKafkaProducer::start()
         return ret;
     }
     
-    std::string enabled = srs_bool2switch(_srs_config->get_kafka_enabled());
-    srs_trace("kafka worker ok, enabled:%s", enabled.c_str());
+    srs_info("kafka worker ok");
     
     return ret;
 }
@@ -80,7 +86,26 @@ int SrsKafkaProducer::request_metadata()
 {
     int ret = ERROR_SUCCESS;
     
-    srs_info("update kafka metadata ok");
+    bool enabled = _srs_config->get_kafka_enabled();
+    if (!enabled) {
+        return ret;
+    }
+    
+    SrsConfDirective* brokers = _srs_config->get_kafka_brokers();
+    if (!brokers) {
+        srs_warn("ignore for empty brokers.");
+        return ret;
+    }
+    
+    srs_assert(!brokers->args.empty());
+    std::string broker = lb->select<string>(brokers->args);
+    
+    if (true) {
+        std::string senabled = srs_bool2switch(enabled);
+        std::string sbrokers = srs_join_vector_string(brokers->args, ",");
+        srs_trace("kafka ok, enabled:%s, brokers:%s, current:[%d]%s",
+            senabled.c_str(), sbrokers.c_str(), lb->current(), broker.c_str());
+    }
     
     return ret;
 }

trunk/src/app/srs_app_kafka.hpp

@@ -29,6 +29,7 @@
 */
 #include <srs_core.hpp>
 
+class SrsLbRoundRobin;
 class SrsAsyncCallWorker;
 
 #ifdef SRS_AUTO_KAFKA
@@ -39,6 +40,7 @@ class SrsAsyncCallWorker;
 class SrsKafkaProducer
 {
 private:
+    SrsLbRoundRobin* lb;
     SrsAsyncCallWorker* worker;
 public:
     SrsKafkaProducer();

trunk/src/kernel/srs_kernel_balance.cpp

@@ -0,0 +1,40 @@
+/*
+ The MIT License (MIT)
+ 
+ Copyright (c) 2013-2015 SRS(simple-rtmp-server)
+ 
+ Permission is hereby granted, free of charge, to any person obtaining a copy of
+ this software and associated documentation files (the "Software"), to deal in
+ the Software without restriction, including without limitation the rights to
+ use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of
+ the Software, and to permit persons to whom the Software is furnished to do so,
+ subject to the following conditions:
+ 
+ The above copyright notice and this permission notice shall be included in all
+ copies or substantial portions of the Software.
+ 
+ THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+ IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS
+ FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR
+ COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER
+ IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN
+ CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
+ */
+
+#include <srs_kernel_balance.hpp>
+
+SrsLbRoundRobin::SrsLbRoundRobin()
+{
+    index = -1;
+    count = 0;
+}
+
+SrsLbRoundRobin::~SrsLbRoundRobin()
+{
+}
+
+u_int32_t SrsLbRoundRobin::current()
+{
+    return index;
+}
+

trunk/src/kernel/srs_kernel_balance.hpp

@@ -0,0 +1,63 @@
+/*
+ The MIT License (MIT)
+ 
+ Copyright (c) 2013-2015 SRS(simple-rtmp-server)
+ 
+ Permission is hereby granted, free of charge, to any person obtaining a copy of
+ this software and associated documentation files (the "Software"), to deal in
+ the Software without restriction, including without limitation the rights to
+ use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of
+ the Software, and to permit persons to whom the Software is furnished to do so,
+ subject to the following conditions:
+ 
+ The above copyright notice and this permission notice shall be included in all
+ copies or substantial portions of the Software.
+ 
+ THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+ IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS
+ FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR
+ COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER
+ IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN
+ CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
+ */
+
+#ifndef SRS_KERNEL_AAC_HPP
+#define SRS_KERNEL_AAC_HPP
+
+/*
+#include <srs_kernel_balance.hpp>
+*/
+#include <srs_core.hpp>
+
+#include <vector>
+
+/**
+ * the round-robin load balance algorithm,
+ * used for edge pull, kafka and other multiple server feature.
+ */
+class SrsLbRoundRobin
+{
+private:
+    // current selected index.
+    int index;
+    // total scheduled count.
+    u_int32_t count;
+public:
+    SrsLbRoundRobin();
+    virtual ~SrsLbRoundRobin();
+public:
+    virtual u_int32_t current();
+public:
+    template<typename T>
+    const T& select(const std::vector<T>& servers)
+    {
+        srs_assert(!servers.empty());
+        
+        index = (int)(count++ % servers.size());
+        
+        return servers.at(index);
+    }
+};
+
+#endif
+