refine mpegts code, use simple rtmp client

@@ -214,8 +214,13 @@ int SrsDynamicHttpConn::do_proxy(ISrsHttpResponseReader* rr, SrsFlvDecoder* dec)
             return ret;
             return ret;
         }
         }
         
         
+        SrsSharedPtrMessage* msg = NULL;
+        if ((ret = sdk->rtmp_create_msg(type, time, data, size, &msg)) != ERROR_SUCCESS) {
+            return ret;
+        }
+        
         // TODO: FIXME: for post flv, reconnect when error.
         // TODO: FIXME: for post flv, reconnect when error.
-        if ((ret = sdk->rtmp_write_packet(type, time, data, size)) != ERROR_SUCCESS) {
+        if ((ret = sdk->send_and_free_message(msg)) != ERROR_SUCCESS) {
             if (!srs_is_client_gracefully_close(ret)) {
             if (!srs_is_client_gracefully_close(ret)) {
                 srs_error("flv: proxy rtmp packet failed. ret=%d", ret);
                 srs_error("flv: proxy rtmp packet failed. ret=%d", ret);
             }
             }

@@ -49,6 +49,7 @@ using namespace std;
 #include <srs_protocol_amf0.hpp>
 #include <srs_protocol_amf0.hpp>
 #include <srs_raw_avc.hpp>
 #include <srs_raw_avc.hpp>
 #include <srs_app_pithy_print.hpp>
 #include <srs_app_pithy_print.hpp>
+#include <srs_app_rtmp_conn.hpp>
 
 
 SrsMpegtsQueue::SrsMpegtsQueue()
 SrsMpegtsQueue::SrsMpegtsQueue()
 {
 {
@@ -132,9 +133,7 @@ SrsMpegtsOverUdp::SrsMpegtsOverUdp(SrsConfDirective* c)
     output = _srs_config->get_stream_caster_output(c);
     output = _srs_config->get_stream_caster_output(c);
     
     
     req = NULL;
     req = NULL;
-    client = NULL;
-    transport = new SrsTcpClient();
-    stream_id = 0;
+    sdk = new SrsSimpleRtmpClient();
     
     
     avc = new SrsRawH264Stream();
     avc = new SrsRawH264Stream();
     aac = new SrsRawAacStream();
     aac = new SrsRawAacStream();
@@ -149,7 +148,7 @@ SrsMpegtsOverUdp::~SrsMpegtsOverUdp()
 {
 {
     close();
     close();
 
 
-    srs_freep(transport);
+    srs_freep(sdk);
     srs_freep(buffer);
     srs_freep(buffer);
     srs_freep(stream);
     srs_freep(stream);
     srs_freep(context);
     srs_freep(context);
@@ -566,7 +565,7 @@ int SrsMpegtsOverUdp::rtmp_write_packet(char type, u_int32_t timestamp, char* da
     
     
     SrsSharedPtrMessage* msg = NULL;
     SrsSharedPtrMessage* msg = NULL;
     
     
-    if ((ret = srs_rtmp_create_msg(type, timestamp, data, size, stream_id, &msg)) != ERROR_SUCCESS) {
+    if ((ret = sdk->rtmp_create_msg(type, timestamp, data, size, &msg)) != ERROR_SUCCESS) {
         srs_error("mpegts: create shared ptr msg failed. ret=%d", ret);
         srs_error("mpegts: create shared ptr msg failed. ret=%d", ret);
         return ret;
         return ret;
     }
     }
@@ -590,7 +589,7 @@ int SrsMpegtsOverUdp::rtmp_write_packet(char type, u_int32_t timestamp, char* da
         }
         }
     
     
         // send out encoded msg.
         // send out encoded msg.
-        if ((ret = client->send_and_free_message(msg, stream_id)) != ERROR_SUCCESS) {
+        if ((ret = sdk->send_and_free_message(msg)) != ERROR_SUCCESS) {
             return ret;
             return ret;
         }
         }
     }
     }
@@ -604,98 +603,19 @@ int SrsMpegtsOverUdp::connect()
 
 
     // when ok, ignore.
     // when ok, ignore.
     // TODO: FIXME: should reconnect when disconnected.
     // TODO: FIXME: should reconnect when disconnected.
-    if (transport->connected()) {
+    if (sdk->connected()) {
         return ret;
         return ret;
     }
     }
     
     
-    // parse uri
-    if (!req) {
-        req = new SrsRequest();
-        srs_parse_rtmp_url(output, req->tcUrl, req->stream);
-        srs_discovery_tc_url(req->tcUrl, req->schema, req->host, req->vhost, req->app, req->port, req->param);
-    }
-
-    // connect host.
-    if ((ret = transport->connect(req->host, req->port, ST_UTIME_NO_TIMEOUT)) != ERROR_SUCCESS) {
-        srs_error("mpegts: connect server %s:%d failed. ret=%d", req->host.c_str(), req->port, ret);
-        return ret;
-    }
-    
-    srs_freep(client);
-    client = new SrsRtmpClient(transport);
-
-    client->set_recv_timeout(SRS_CONSTS_RTMP_TIMEOUT_US);
-    client->set_send_timeout(SRS_CONSTS_RTMP_TIMEOUT_US);
-    
-    // connect to vhost/app
-    if ((ret = client->handshake()) != ERROR_SUCCESS) {
-        srs_error("mpegts: handshake with server failed. ret=%d", ret);
-        return ret;
-    }
-    if ((ret = connect_app(req->host, req->port)) != ERROR_SUCCESS) {
-        srs_error("mpegts: connect with server failed. ret=%d", ret);
-        return ret;
-    }
-    if ((ret = client->create_stream(stream_id)) != ERROR_SUCCESS) {
-        srs_error("mpegts: connect with server failed, stream_id=%d. ret=%d", stream_id, ret);
+    int64_t cto = SRS_CONSTS_RTMP_TIMEOUT_US;
+    int64_t sto = SRS_CONSTS_RTMP_PULSE_TIMEOUT_US;
+    if ((ret = sdk->connect(output, cto, sto)) != ERROR_SUCCESS) {
+        srs_error("mpegts: connect %s failed, cto=%"PRId64", sto=%"PRId64". ret=%d", output.c_str(), cto, sto, ret);
         return ret;
         return ret;
     }
     }
     
     
-    // publish.
-    if ((ret = client->publish(req->stream, stream_id)) != ERROR_SUCCESS) {
-        srs_error("mpegts: publish failed, stream=%s, stream_id=%d. ret=%d", 
-            req->stream.c_str(), stream_id, ret);
-        return ret;
-    }
-
-    return ret;
-}
-
-// TODO: FIXME: refine the connect_app.
-int SrsMpegtsOverUdp::connect_app(string ep_server, int ep_port)
-{
-    int ret = ERROR_SUCCESS;
-    
-    // args of request takes the srs info.
-    if (req->args == NULL) {
-        req->args = SrsAmf0Any::object();
-    }
-    
-    // notify server the edge identity,
-    // @see https://github.com/simple-rtmp-server/srs/issues/147
-    SrsAmf0Object* data = req->args;
-    data->set("srs_sig", SrsAmf0Any::str(RTMP_SIG_SRS_KEY));
-    data->set("srs_server", SrsAmf0Any::str(RTMP_SIG_SRS_KEY" "RTMP_SIG_SRS_VERSION" ("RTMP_SIG_SRS_URL_SHORT")"));
-    data->set("srs_license", SrsAmf0Any::str(RTMP_SIG_SRS_LICENSE));
-    data->set("srs_role", SrsAmf0Any::str(RTMP_SIG_SRS_ROLE));
-    data->set("srs_url", SrsAmf0Any::str(RTMP_SIG_SRS_URL));
-    data->set("srs_version", SrsAmf0Any::str(RTMP_SIG_SRS_VERSION));
-    data->set("srs_site", SrsAmf0Any::str(RTMP_SIG_SRS_WEB));
-    data->set("srs_email", SrsAmf0Any::str(RTMP_SIG_SRS_EMAIL));
-    data->set("srs_copyright", SrsAmf0Any::str(RTMP_SIG_SRS_COPYRIGHT));
-    data->set("srs_primary", SrsAmf0Any::str(RTMP_SIG_SRS_PRIMARY));
-    data->set("srs_authors", SrsAmf0Any::str(RTMP_SIG_SRS_AUTHROS));
-    // for edge to directly get the id of client.
-    data->set("srs_pid", SrsAmf0Any::number(getpid()));
-    data->set("srs_id", SrsAmf0Any::number(_srs_context->get_id()));
-    
-    // local ip of edge
-    std::vector<std::string> ips = srs_get_local_ipv4_ips();
-    assert(_srs_config->get_stats_network() < (int)ips.size());
-    std::string local_ip = ips[_srs_config->get_stats_network()];
-    data->set("srs_server_ip", SrsAmf0Any::str(local_ip.c_str()));
-    
-    // generate the tcUrl
-    std::string param = "";
-    std::string tc_url = srs_generate_tc_url(ep_server, req->vhost, req->app, ep_port, param);
-    
-    // upnode server identity will show in the connect_app of client.
-    // @see https://github.com/simple-rtmp-server/srs/issues/160
-    // the debug_srs_upnode is config in vhost and default to true.
-    bool debug_srs_upnode = _srs_config->get_debug_srs_upnode(req->vhost);
-    if ((ret = client->connect_app(req->app, tc_url, req, debug_srs_upnode)) != ERROR_SUCCESS) {
-        srs_error("mpegts: connect with server failed, tcUrl=%s, dsu=%d. ret=%d", 
-            tc_url.c_str(), debug_srs_upnode, ret);
+    if ((ret = sdk->publish()) != ERROR_SUCCESS) {
+        srs_error("mpegts: publish failed. ret=%d", ret);
         return ret;
         return ret;
     }
     }
 
 
@@ -704,10 +624,8 @@ int SrsMpegtsOverUdp::connect_app(string ep_server, int ep_port)
 
 
 void SrsMpegtsOverUdp::close()
 void SrsMpegtsOverUdp::close()
 {
 {
-    srs_freep(client);
     srs_freep(req);
     srs_freep(req);
-    
-    transport->close();
+    sdk->close();
 }
 }
 
 
 #endif
 #endif

@@ -48,6 +48,7 @@ class SrsSharedPtrMessage;
 class SrsRawAacStream;
 class SrsRawAacStream;
 struct SrsRawAacStreamCodec;
 struct SrsRawAacStreamCodec;
 class SrsPithyPrint;
 class SrsPithyPrint;
+class SrsSimpleRtmpClient;
 
 
 #include <srs_app_st.hpp>
 #include <srs_app_st.hpp>
 #include <srs_kernel_ts.hpp>
 #include <srs_kernel_ts.hpp>
@@ -86,9 +87,7 @@ private:
     std::string output;
     std::string output;
 private:
 private:
     SrsRequest* req;
     SrsRequest* req;
-    SrsTcpClient* transport;
-    SrsRtmpClient* client;
-    int stream_id;
+    SrsSimpleRtmpClient* sdk;
 private:
 private:
     SrsRawH264Stream* avc;
     SrsRawH264Stream* avc;
     std::string h264_sps;
     std::string h264_sps;
@@ -125,7 +124,6 @@ private:
     // connect to rtmp output url. 
     // connect to rtmp output url. 
     // @remark ignore when not connected, reconnect when disconnected.
     // @remark ignore when not connected, reconnect when disconnected.
     virtual int connect();
     virtual int connect();
-    virtual int connect_app(std::string ep_server, int ep_port);
     // close the connected io and rtmp to ready to be re-connect.
     // close the connected io and rtmp to ready to be re-connect.
     virtual void close();
     virtual void close();
 };
 };

@@ -198,6 +198,11 @@ int SrsSimpleRtmpClient::connect_app()
     return ret;
     return ret;
 }
 }
 
 
+bool SrsSimpleRtmpClient::connected()
+{
+    return transport->connected();
+}
+
 void SrsSimpleRtmpClient::close()
 void SrsSimpleRtmpClient::close()
 {
 {
     transport->close();
     transport->close();
@@ -266,22 +271,16 @@ int SrsSimpleRtmpClient::sid()
     return stream_id;
     return stream_id;
 }
 }
 
 
-int SrsSimpleRtmpClient::rtmp_write_packet(char type, u_int32_t timestamp, char* data, int size)
+int SrsSimpleRtmpClient::rtmp_create_msg(char type, u_int32_t timestamp, char* data, int size, SrsSharedPtrMessage** pmsg)
 {
 {
-    int ret = ERROR_SUCCESS;
+    *pmsg = NULL;
     
     
-    SrsSharedPtrMessage* msg = NULL;
+    int ret = ERROR_SUCCESS;
     
     
-    if ((ret = srs_rtmp_create_msg(type, timestamp, data, size, stream_id, &msg)) != ERROR_SUCCESS) {
+    if ((ret = srs_rtmp_create_msg(type, timestamp, data, size, stream_id, pmsg)) != ERROR_SUCCESS) {
         srs_error("sdk: create shared ptr msg failed. ret=%d", ret);
         srs_error("sdk: create shared ptr msg failed. ret=%d", ret);
         return ret;
         return ret;
     }
     }
-    srs_assert(msg);
-    
-    // send out encoded msg.
-    if ((ret = client->send_and_free_message(msg, stream_id)) != ERROR_SUCCESS) {
-        return ret;
-    }
     
     
     return ret;
     return ret;
 }
 }

@@ -78,6 +78,7 @@ public:
 private:
 private:
     virtual int connect_app();
     virtual int connect_app();
 public:
 public:
+    virtual bool connected();
     virtual void close();
     virtual void close();
 public:
 public:
     virtual int publish();
     virtual int publish();
@@ -86,7 +87,8 @@ public:
     virtual void kbps_sample(const char* label, int64_t age, int msgs);
     virtual void kbps_sample(const char* label, int64_t age, int msgs);
     virtual int sid();
     virtual int sid();
 public:
 public:
-    virtual int rtmp_write_packet(char type, u_int32_t timestamp, char* data, int size);
+    virtual int rtmp_create_msg(char type, u_int32_t timestamp, char* data, int size, SrsSharedPtrMessage** pmsg);
+public:
     virtual int recv_message(SrsCommonMessage** pmsg);
     virtual int recv_message(SrsCommonMessage** pmsg);
     virtual int decode_message(SrsCommonMessage* msg, SrsPacket** ppacket);
     virtual int decode_message(SrsCommonMessage* msg, SrsPacket** ppacket);
     virtual int send_and_free_messages(SrsSharedPtrMessage** msgs, int nb_msgs);
     virtual int send_and_free_messages(SrsSharedPtrMessage** msgs, int nb_msgs);