For #464, support origin cluster api

trunk/configure

@@ -200,7 +200,8 @@ if [ $SRS_EXPORT_LIBRTMP_PROJECT = NO ]; then
             "srs_app_recv_thread" "srs_app_security" "srs_app_statistic" "srs_app_hds"
             "srs_app_mpegts_udp" "srs_app_rtsp" "srs_app_listener" "srs_app_async_call"
             "srs_app_caster_flv" "srs_app_process" "srs_app_ng_exec" "srs_app_kafka"
-            "srs_app_hourglass" "srs_app_dash" "srs_app_fragment" "srs_app_dvr")
+            "srs_app_hourglass" "srs_app_dash" "srs_app_fragment" "srs_app_dvr"
+            "srs_app_coworkers")
     DEFINES=""
     # add each modules for app
     for SRS_MODULE in ${SRS_MODULES[*]}; do

trunk/ide/srs_xcode/srs_xcode.xcodeproj/project.pbxproj

@@ -132,6 +132,7 @@
 		3CECAF9E1EDC100F00C50501 /* sched.c in Sources */ = {isa = PBXBuildFile; fileRef = 3CECAF961EDC100F00C50501 /* sched.c */; };
 		3CECAF9F1EDC100F00C50501 /* stk.c in Sources */ = {isa = PBXBuildFile; fileRef = 3CECAF971EDC100F00C50501 /* stk.c */; };
 		3CECAFA01EDC100F00C50501 /* sync.c in Sources */ = {isa = PBXBuildFile; fileRef = 3CECAF981EDC100F00C50501 /* sync.c */; };
+		8C0652B12035B5BA000B0661 /* srs_app_coworkers.cpp in Sources */ = {isa = PBXBuildFile; fileRef = 8C0652B02035B5B9000B0661 /* srs_app_coworkers.cpp */; };
 /* End PBXBuildFile section */
 
 /* Begin PBXCopyFilesBuildPhase section */
@@ -446,6 +447,8 @@
 		3CECAF971EDC100F00C50501 /* stk.c */ = {isa = PBXFileReference; fileEncoding = 4; lastKnownFileType = sourcecode.c.c; name = stk.c; path = "../../../objs/state-threads-1.9.1/stk.c"; sourceTree = "<group>"; };
 		3CECAF981EDC100F00C50501 /* sync.c */ = {isa = PBXFileReference; fileEncoding = 4; lastKnownFileType = sourcecode.c.c; name = sync.c; path = "../../../objs/state-threads-1.9.1/sync.c"; sourceTree = "<group>"; };
 		3CFBDA271F0338A40054D63E /* circle.yml */ = {isa = PBXFileReference; fileEncoding = 4; lastKnownFileType = text; name = circle.yml; path = ../../../circle.yml; sourceTree = "<group>"; };
+		8C0652AF2035B5B9000B0661 /* srs_app_coworkers.hpp */ = {isa = PBXFileReference; fileEncoding = 4; lastKnownFileType = sourcecode.cpp.h; name = srs_app_coworkers.hpp; path = ../../../src/app/srs_app_coworkers.hpp; sourceTree = "<group>"; };
+		8C0652B02035B5B9000B0661 /* srs_app_coworkers.cpp */ = {isa = PBXFileReference; fileEncoding = 4; lastKnownFileType = sourcecode.cpp.cpp; name = srs_app_coworkers.cpp; path = ../../../src/app/srs_app_coworkers.cpp; sourceTree = "<group>"; };
 /* End PBXFileReference section */
 
 /* Begin PBXFrameworksBuildPhase section */
@@ -622,6 +625,8 @@
 				3C12324F1AAE81D900CE8F6C /* srs_app_config.hpp */,
 				3C1232501AAE81D900CE8F6C /* srs_app_conn.cpp */,
 				3C1232511AAE81D900CE8F6C /* srs_app_conn.hpp */,
+				8C0652B02035B5B9000B0661 /* srs_app_coworkers.cpp */,
+				8C0652AF2035B5B9000B0661 /* srs_app_coworkers.hpp */,
 				3C9F82201E4ECA8200F5B2D2 /* srs_app_dash.cpp */,
 				3C9F82211E4ECA8200F5B2D2 /* srs_app_dash.hpp */,
 				3CE893B31E87508D000B742D /* srs_app_dvr.cpp */,
@@ -1056,6 +1061,7 @@
 				3CC52DDC1ACE4023006FEB01 /* srs_utest_protocol.cpp in Sources */,
 				3C663F151AB0155100286D8B /* srs_h264_raw_publish.c in Sources */,
 				3C1231F61AAE652D00CE8F6C /* srs_core_autofree.cpp in Sources */,
+				8C0652B12035B5BA000B0661 /* srs_app_coworkers.cpp in Sources */,
 				3C5265B41B241BF0009CA186 /* srs_core_mem_watch.cpp in Sources */,
 				3C1EE6D71AB1367D00576EE9 /* README.md in Sources */,
 				3C82802C1BAFF8CC004A1794 /* srs_kafka_stack.cpp in Sources */,

trunk/src/app/srs_app_coworkers.cpp

@@ -0,0 +1,63 @@
+/**
+ * The MIT License (MIT)
+ *
+ * Copyright (c) 2013-2018 Winlin
+ *
+ * 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_app_coworkers.hpp>
+
+using namespace std;
+
+#include <srs_protocol_json.hpp>
+#include <srs_kernel_error.hpp>
+
+SrsCoWorkers* SrsCoWorkers::_instance = NULL;
+
+SrsCoWorkers::SrsCoWorkers()
+{
+}
+
+SrsCoWorkers::~SrsCoWorkers()
+{
+}
+
+SrsCoWorkers* SrsCoWorkers::instance()
+{
+    if (!_instance) {
+        _instance = new SrsCoWorkers();
+    }
+    return _instance;
+}
+
+SrsJsonAny* SrsCoWorkers::dumps(string vhost, string app, string stream)
+{
+    return SrsJsonAny::null();
+}
+
+srs_error_t SrsCoWorkers::on_publish(SrsSource* s, SrsRequest* r)
+{
+    srs_error_t err = srs_success;
+    return err;
+}
+
+void SrsCoWorkers::on_unpublish(SrsSource* s, SrsRequest* r)
+{
+}
+

trunk/src/app/srs_app_coworkers.hpp

@@ -0,0 +1,51 @@
+/**
+ * The MIT License (MIT)
+ *
+ * Copyright (c) 2013-2018 Winlin
+ *
+ * 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_APP_COWORKERS_HPP
+#define SRS_APP_COWORKERS_HPP
+
+#include <srs_core.hpp>
+
+#include <string>
+
+class SrsJsonAny;
+class SrsRequest;
+class SrsSource;
+
+class SrsCoWorkers
+{
+private:
+    static SrsCoWorkers* _instance;
+private:
+    SrsCoWorkers();
+    virtual ~SrsCoWorkers();
+public:
+    static SrsCoWorkers* instance();
+public:
+    virtual SrsJsonAny* dumps(std::string vhost, std::string app, std::string stream);
+public:
+    virtual srs_error_t on_publish(SrsSource* s, SrsRequest* r);
+    virtual void on_unpublish(SrsSource* s, SrsRequest* r);
+};
+
+#endif

trunk/src/app/srs_app_http_api.cpp

@@ -45,6 +45,7 @@ using namespace std;
 #include <srs_app_server.hpp>
 #include <srs_protocol_amf0.hpp>
 #include <srs_protocol_utility.hpp>
+#include <srs_app_coworkers.hpp>
 
 srs_error_t srs_api_response_jsonp(ISrsHttpResponseWriter* w, string callback, string data)
 {
@@ -262,6 +263,7 @@ srs_error_t SrsGoApiV1::serve_http(ISrsHttpResponseWriter* w, ISrsHttpMessage* r
     urls->set("streams", SrsJsonAny::str("manage all streams or specified stream"));
     urls->set("clients", SrsJsonAny::str("manage all clients or specified client, default query top 10 clients"));
     urls->set("raw", SrsJsonAny::str("raw api for srs, support CUID srs for instance the config"));
+    urls->set("clusters", SrsJsonAny::str("origin cluster server API"));
     
     SrsJsonObject* tests = SrsJsonAny::object();
     obj->set("tests", tests);
@@ -1290,6 +1292,39 @@ srs_error_t SrsGoApiRaw::on_reload_http_api_raw_api()
     return srs_success;
 }
 
+SrsGoApiClusters::SrsGoApiClusters()
+{
+}
+
+SrsGoApiClusters::~SrsGoApiClusters()
+{
+}
+
+srs_error_t SrsGoApiClusters::serve_http(ISrsHttpResponseWriter* w, ISrsHttpMessage* r)
+{
+    SrsJsonObject* obj = SrsJsonAny::object();
+    SrsAutoFree(SrsJsonObject, obj);
+    
+    obj->set("code", SrsJsonAny::integer(ERROR_SUCCESS));
+    SrsJsonObject* data = SrsJsonAny::object();
+    obj->set("data", data);
+    
+    string ip = r->query_get("ip");
+    string vhost = r->query_get("vhost");
+    string app = r->query_get("app");
+    string stream = r->query_get("stream");
+    data->set("query", SrsJsonAny::object()
+              ->set("ip", SrsJsonAny::str(ip.c_str()))
+              ->set("vhost", SrsJsonAny::str(vhost.c_str()))
+              ->set("app", SrsJsonAny::str(app.c_str()))
+              ->set("stream", SrsJsonAny::str(stream.c_str())));
+    
+    SrsCoWorkers* coworkers = SrsCoWorkers::instance();
+    data->set("origin", coworkers->dumps(vhost, app, stream));
+    
+    return srs_api_response(w, r, obj->dumps());
+}
+
 SrsGoApiError::SrsGoApiError()
 {
 }

trunk/src/app/srs_app_http_api.hpp

@@ -192,6 +192,15 @@ public:
     virtual srs_error_t on_reload_http_api_raw_api();
 };
 
+class SrsGoApiClusters : public ISrsHttpHandler
+{
+public:
+    SrsGoApiClusters();
+    virtual ~SrsGoApiClusters();
+public:
+    virtual srs_error_t serve_http(ISrsHttpResponseWriter* w, ISrsHttpMessage* r);
+};
+
 class SrsGoApiError : public ISrsHttpHandler
 {
 public:

trunk/src/app/srs_app_server.cpp

@@ -51,6 +51,7 @@ using namespace std;
 #include <srs_kernel_consts.hpp>
 #include <srs_app_kafka.hpp>
 #include <srs_app_thread.hpp>
+#include <srs_app_coworkers.hpp>
 
 // system interval in ms,
 // all resolution times should be times togother,
@@ -772,6 +773,9 @@ srs_error_t SrsServer::http_handle()
     if ((err = http_api_mux->handle("/api/v1/raw", new SrsGoApiRaw(this))) != srs_success) {
         return srs_error_wrap(err, "handle raw");
     }
+    if ((err = http_api_mux->handle("/api/v1/clusters", new SrsGoApiClusters())) != srs_success) {
+        return srs_error_wrap(err, "handle raw");
+    }
     
     // test the request info.
     if ((err = http_api_mux->handle("/api/v1/tests/requests", new SrsGoApiRequests())) != srs_success) {
@@ -1397,11 +1401,19 @@ srs_error_t SrsServer::on_publish(SrsSource* s, SrsRequest* r)
         return srs_error_wrap(err, "http mount");
     }
     
+    SrsCoWorkers* coworkers = SrsCoWorkers::instance();
+    if ((err = coworkers->on_publish(s, r)) != srs_success) {
+        return srs_error_wrap(err, "coworkers");
+    }
+    
     return err;
 }
 
 void SrsServer::on_unpublish(SrsSource* s, SrsRequest* r)
 {
     http_server->http_unmount(s, r);
+    
+    SrsCoWorkers* coworkers = SrsCoWorkers::instance();
+    coworkers->on_unpublish(s, r);
 }
 

trunk/src/protocol/srs_protocol_json.cpp

@@ -1818,11 +1818,11 @@ SrsAmf0Any* SrsJsonObject::to_amf0()
     return obj;
 }
 
-void SrsJsonObject::set(string key, SrsJsonAny* value)
+SrsJsonObject* SrsJsonObject::set(string key, SrsJsonAny* value)
 {
     if (!value) {
         srs_warn("add a NULL propertity %s", key.c_str());
-        return;
+        return this;
     }
     
     std::vector<SrsJsonObjectPropertyType>::iterator it;
@@ -1840,6 +1840,7 @@ void SrsJsonObject::set(string key, SrsJsonAny* value)
     }
     
     properties.push_back(std::make_pair(key, value));
+    return this;
 }
 
 SrsJsonAny* SrsJsonObject::get_property(string name)
@@ -1980,9 +1981,10 @@ void SrsJsonArray::add(SrsJsonAny* value)
     properties.push_back(value);
 }
 
-void SrsJsonArray::append(SrsJsonAny* value)
+SrsJsonArray* SrsJsonArray::append(SrsJsonAny* value)
 {
     add(value);
+    return this;
 }
 
 string SrsJsonArray::dumps()

trunk/src/protocol/srs_protocol_json.hpp

@@ -145,7 +145,7 @@ public:
     virtual std::string dumps();
     virtual SrsAmf0Any* to_amf0();
 public:
-    virtual void set(std::string key, SrsJsonAny* value);
+    virtual SrsJsonObject* set(std::string key, SrsJsonAny* value);
     virtual SrsJsonAny* get_property(std::string name);
     virtual SrsJsonAny* ensure_property_string(std::string name);
     virtual SrsJsonAny* ensure_property_integer(std::string name);
@@ -172,7 +172,7 @@ public:
     virtual SrsJsonAny* at(int index);
     virtual void add(SrsJsonAny* value);
     // alias to add.
-    virtual void append(SrsJsonAny* value);
+    virtual SrsJsonArray* append(SrsJsonAny* value);
 public:
     virtual std::string dumps();
     virtual SrsAmf0Any* to_amf0();