diff --git a/trunk/src/app/srs_app_caster_flv.cpp b/trunk/src/app/srs_app_caster_flv.cpp index 5da82e8fe..f8258f896 100644 --- a/trunk/src/app/srs_app_caster_flv.cpp +++ b/trunk/src/app/srs_app_caster_flv.cpp @@ -214,8 +214,13 @@ int SrsDynamicHttpConn::do_proxy(ISrsHttpResponseReader* rr, SrsFlvDecoder* dec) 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. - 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)) { srs_error("flv: proxy rtmp packet failed. ret=%d", ret); } diff --git a/trunk/src/app/srs_app_mpegts_udp.cpp b/trunk/src/app/srs_app_mpegts_udp.cpp index 5f1497615..0d437166c 100644 --- a/trunk/src/app/srs_app_mpegts_udp.cpp +++ b/trunk/src/app/srs_app_mpegts_udp.cpp @@ -49,6 +49,7 @@ using namespace std; #include #include #include +#include SrsMpegtsQueue::SrsMpegtsQueue() { @@ -132,9 +133,7 @@ SrsMpegtsOverUdp::SrsMpegtsOverUdp(SrsConfDirective* c) output = _srs_config->get_stream_caster_output(c); req = NULL; - client = NULL; - transport = new SrsTcpClient(); - stream_id = 0; + sdk = new SrsSimpleRtmpClient(); avc = new SrsRawH264Stream(); aac = new SrsRawAacStream(); @@ -149,7 +148,7 @@ SrsMpegtsOverUdp::~SrsMpegtsOverUdp() { close(); - srs_freep(transport); + srs_freep(sdk); srs_freep(buffer); srs_freep(stream); srs_freep(context); @@ -565,8 +564,8 @@ int SrsMpegtsOverUdp::rtmp_write_packet(char type, u_int32_t timestamp, char* da int ret = ERROR_SUCCESS; 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); return ret; } @@ -590,7 +589,7 @@ int SrsMpegtsOverUdp::rtmp_write_packet(char type, u_int32_t timestamp, char* da } // 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; } } @@ -604,110 +603,29 @@ int SrsMpegtsOverUdp::connect() // when ok, ignore. // TODO: FIXME: should reconnect when disconnected. - if (transport->connected()) { + if (sdk->connected()) { 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); + 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; } - 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); - 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); + if ((ret = sdk->publish()) != ERROR_SUCCESS) { + srs_error("mpegts: publish failed. ret=%d", 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 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); - return ret; - } - - return ret; -} - void SrsMpegtsOverUdp::close() { - srs_freep(client); srs_freep(req); - - transport->close(); + sdk->close(); } #endif diff --git a/trunk/src/app/srs_app_mpegts_udp.hpp b/trunk/src/app/srs_app_mpegts_udp.hpp index f2ba65727..e9de14c00 100644 --- a/trunk/src/app/srs_app_mpegts_udp.hpp +++ b/trunk/src/app/srs_app_mpegts_udp.hpp @@ -48,6 +48,7 @@ class SrsSharedPtrMessage; class SrsRawAacStream; struct SrsRawAacStreamCodec; class SrsPithyPrint; +class SrsSimpleRtmpClient; #include #include @@ -86,9 +87,7 @@ private: std::string output; private: SrsRequest* req; - SrsTcpClient* transport; - SrsRtmpClient* client; - int stream_id; + SrsSimpleRtmpClient* sdk; private: SrsRawH264Stream* avc; std::string h264_sps; @@ -125,7 +124,6 @@ private: // connect to rtmp output url. // @remark ignore when not connected, reconnect when disconnected. 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. virtual void close(); }; diff --git a/trunk/src/app/srs_app_rtmp_conn.cpp b/trunk/src/app/srs_app_rtmp_conn.cpp index 0301c4602..a9229611a 100644 --- a/trunk/src/app/srs_app_rtmp_conn.cpp +++ b/trunk/src/app/srs_app_rtmp_conn.cpp @@ -198,6 +198,11 @@ int SrsSimpleRtmpClient::connect_app() return ret; } +bool SrsSimpleRtmpClient::connected() +{ + return transport->connected(); +} + void SrsSimpleRtmpClient::close() { transport->close(); @@ -266,22 +271,16 @@ int SrsSimpleRtmpClient::sid() 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) { + *pmsg = NULL; + int ret = ERROR_SUCCESS; - SrsSharedPtrMessage* msg = NULL; - - 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); 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; } diff --git a/trunk/src/app/srs_app_rtmp_conn.hpp b/trunk/src/app/srs_app_rtmp_conn.hpp index be345660d..5cfc0624f 100644 --- a/trunk/src/app/srs_app_rtmp_conn.hpp +++ b/trunk/src/app/srs_app_rtmp_conn.hpp @@ -78,6 +78,7 @@ public: private: virtual int connect_app(); public: + virtual bool connected(); virtual void close(); public: virtual int publish(); @@ -86,7 +87,8 @@ public: virtual void kbps_sample(const char* label, int64_t age, int msgs); virtual int sid(); 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 decode_message(SrsCommonMessage* msg, SrsPacket** ppacket); virtual int send_and_free_messages(SrsSharedPtrMessage** msgs, int nb_msgs);