refine code.

trunk/src/app/srs_app_kafka.cpp

@@ -381,25 +381,37 @@ void SrsKafkaProducer::stop()
     worker->stop();
 }
 
-int SrsKafkaProducer::on_client(int key, SrsListenerType type, string ip)
+int SrsKafkaProducer::send(int key, SrsJsonObject* obj)
 {
     int ret = ERROR_SUCCESS;
     
-    bool enabled = _srs_config->get_kafka_enabled();
+    // cache the json object.
-    if (!enabled) {
+    cache->append(key, obj);
+    
+    // too few messages, ignore.
+    if (cache->size() < SRS_KAFKA_PRODUCER_AGGREGATE_SIZE) {
         return ret;
     }
     
-    SrsJsonObject* obj = SrsJsonAny::object();
+    // too many messages, warn user.
+    if (cache->size() > SRS_KAFKA_PRODUCER_AGGREGATE_SIZE * 10) {
+        srs_warn("kafka cache too many messages: %d", cache->size());
+    }
     
-    obj->set("msg", SrsJsonAny::str("accept"));
+    // sync with backgound metadata worker.
-    obj->set("type", SrsJsonAny::integer(type));
+    st_mutex_lock(lock);
-    obj->set("ip", SrsJsonAny::str(ip.c_str()));
     
-    return worker->execute(new SrsKafkaMessage(this, key, obj));
+    // flush message when metadata is ok.
+    if (metadata_ok) {
+        ret = flush();
+    }
+    
+    st_mutex_unlock(lock);
+    
+    return ret;
 }
 
-int SrsKafkaProducer::on_close(int key)
+int SrsKafkaProducer::on_client(int key, SrsListenerType type, string ip)
 {
     int ret = ERROR_SUCCESS;
     
@@ -410,39 +422,27 @@ int SrsKafkaProducer::on_close(int key)
     
     SrsJsonObject* obj = SrsJsonAny::object();
     
-    obj->set("msg", SrsJsonAny::str("close"));
+    obj->set("msg", SrsJsonAny::str("accept"));
+    obj->set("type", SrsJsonAny::integer(type));
+    obj->set("ip", SrsJsonAny::str(ip.c_str()));
     
     return worker->execute(new SrsKafkaMessage(this, key, obj));
 }
 
-int SrsKafkaProducer::send(int key, SrsJsonObject* obj)
+int SrsKafkaProducer::on_close(int key)
 {
     int ret = ERROR_SUCCESS;
     
-    // cache the json object.
+    bool enabled = _srs_config->get_kafka_enabled();
-    cache->append(key, obj);
+    if (!enabled) {
-    
-    // too few messages, ignore.
-    if (cache->size() < SRS_KAFKA_PRODUCER_AGGREGATE_SIZE) {
         return ret;
     }
     
-    // too many messages, warn user.
+    SrsJsonObject* obj = SrsJsonAny::object();
-    if (cache->size() > SRS_KAFKA_PRODUCER_AGGREGATE_SIZE * 10) {
-        srs_warn("kafka cache too many messages: %d", cache->size());
-    }
-    
-    // sync with backgound metadata worker.
-    st_mutex_lock(lock);
-    
-    // flush message when metadata is ok.
-    if (metadata_ok) {
-        ret = flush();
-    }
     
-    st_mutex_unlock(lock);
+    obj->set("msg", SrsJsonAny::str("close"));
     
-    return ret;
+    return worker->execute(new SrsKafkaMessage(this, key, obj));
 }
 
 int SrsKafkaProducer::cycle()

trunk/src/app/srs_app_kafka.hpp

@@ -173,11 +173,7 @@ public:
     virtual int initialize();
     virtual int start();
     virtual void stop();
-// interface ISrsKafkaCluster
+// internal: for worker to call task to send object.
-public:
-    virtual int on_client(int key, SrsListenerType type, std::string ip);
-    virtual int on_close(int key);
-// for worker to call task to send object.
 public:
     /**
      * send json object to kafka cluster.
@@ -186,6 +182,10 @@ public:
      * @param obj the json object; user must never free it again.
      */
     virtual int send(int key, SrsJsonObject* obj);
+// interface ISrsKafkaCluster
+public:
+    virtual int on_client(int key, SrsListenerType type, std::string ip);
+    virtual int on_close(int key);
 // interface ISrsReusableThreadHandler
 public:
     virtual int cycle();