diff --git a/trunk/src/app/srs_app_dash.cpp b/trunk/src/app/srs_app_dash.cpp index 922c1672b..88eb05ae7 100644 --- a/trunk/src/app/srs_app_dash.cpp +++ b/trunk/src/app/srs_app_dash.cpp @@ -172,19 +172,31 @@ SrsMpdWriter::~SrsMpdWriter() srs_error_t SrsMpdWriter::initialize(SrsRequest* r) { req = r; + return srs_success; +} + +srs_error_t SrsMpdWriter::on_publish() +{ + SrsRequest* r = req; + fragment = _srs_config->get_dash_fragment(r->vhost); update_period = _srs_config->get_dash_update_period(r->vhost); timeshit = _srs_config->get_dash_timeshift(r->vhost); home = _srs_config->get_dash_path(r->vhost); mpd_file = _srs_config->get_dash_mpd_file(r->vhost); - + string mpd_path = srs_path_build_stream(mpd_file, req->vhost, req->app, req->stream); fragment_home = srs_path_dirname(mpd_path) + "/" + req->stream; - + srs_trace("DASH: Config fragment=%" PRId64 ", period=%" PRId64, fragment, update_period); + return srs_success; } +void SrsMpdWriter::on_unpublish() +{ +} + srs_error_t SrsMpdWriter::write(SrsFormat* format) { srs_error_t err = srs_success; @@ -303,8 +315,6 @@ srs_error_t SrsDashController::initialize(SrsRequest* r) srs_error_t err = srs_success; req = r; - fragment = _srs_config->get_dash_fragment(r->vhost); - home = _srs_config->get_dash_path(r->vhost); if ((err = mpd->initialize(r)) != srs_success) { return srs_error_wrap(err, "mpd"); @@ -317,6 +327,11 @@ srs_error_t SrsDashController::on_publish() { srs_error_t err = srs_success; + SrsRequest* r = req; + + fragment = _srs_config->get_dash_fragment(r->vhost); + home = _srs_config->get_dash_path(r->vhost); + srs_freep(vcurrent); vcurrent = new SrsFragmentedMp4(); if ((err = vcurrent->initialize(req, true, mpd, video_tack_id)) != srs_success) { @@ -329,11 +344,17 @@ srs_error_t SrsDashController::on_publish() return srs_error_wrap(err, "audio fragment"); } + if ((err = mpd->on_publish()) != srs_success) { + return srs_error_wrap(err, "mpd"); + } + return err; } void SrsDashController::on_unpublish() { + mpd->on_unpublish(); + srs_freep(vcurrent); srs_freep(acurrent); } diff --git a/trunk/src/app/srs_app_dash.hpp b/trunk/src/app/srs_app_dash.hpp index 51f19bde9..85559802b 100644 --- a/trunk/src/app/srs_app_dash.hpp +++ b/trunk/src/app/srs_app_dash.hpp @@ -103,6 +103,8 @@ public: virtual ~SrsMpdWriter(); public: virtual srs_error_t initialize(SrsRequest* r); + virtual srs_error_t on_publish(); + virtual void on_unpublish(); // Write MPD according to parsed format of stream. virtual srs_error_t write(SrsFormat* format); public: diff --git a/trunk/src/app/srs_app_kafka.cpp b/trunk/src/app/srs_app_kafka.cpp index de5974fc6..2e656b69c 100644 --- a/trunk/src/app/srs_app_kafka.cpp +++ b/trunk/src/app/srs_app_kafka.cpp @@ -40,7 +40,7 @@ using namespace std; #ifdef SRS_AUTO_KAFKA -#define SRS_KAFKA_PRODUCER_TIMEOUT 30000 +#define SRS_KAFKA_PRODUCER_TIMEOUT (30 * SRS_UTIME_MILLISECONDS) #define SRS_KAFKA_PRODUCER_AGGREGATE_SIZE 1 std::string srs_kafka_metadata_summary(SrsKafkaTopicMetadataResponse* metadata) @@ -577,7 +577,7 @@ srs_error_t SrsKafkaProducer::request_metadata() senabled.c_str(), sbrokers.c_str(), lb->current(), server.c_str(), port, topic.c_str()); } - SrsTcpClient* transport = new SrsTcpClient(server, port, SRS_CONSTS_KAFKA_TMMS); + SrsTcpClient* transport = new SrsTcpClient(server, port, SRS_CONSTS_KAFKA_TIMEOUT); SrsAutoFree(SrsTcpClient, transport); SrsKafkaClient* kafka = new SrsKafkaClient(transport); diff --git a/trunk/src/app/srs_app_rtmp_conn.cpp b/trunk/src/app/srs_app_rtmp_conn.cpp index 886174841..7c4ad4710 100644 --- a/trunk/src/app/srs_app_rtmp_conn.cpp +++ b/trunk/src/app/srs_app_rtmp_conn.cpp @@ -75,7 +75,7 @@ using namespace std; #define SRS_PAUSED_RECV_TMMS (3 * SRS_UTIME_MINUTES) // when edge timeout, retry next. -#define SRS_EDGE_TOKEN_TRAVERSE_TMMS (3000) +#define SRS_EDGE_TOKEN_TRAVERSE_TIMEOUT (3 * SRS_UTIME_SECONDS) SrsSimpleRtmpClient::SrsSimpleRtmpClient(string u, int64_t ctm, int64_t stm) : SrsBasicRtmpClient(u, ctm, stm) { @@ -1159,7 +1159,7 @@ srs_error_t SrsRtmpConn::check_edge_token_traverse_auth() int port = SRS_CONSTS_RTMP_DEFAULT_PORT; srs_parse_hostport(hostport, server, port); - SrsTcpClient* transport = new SrsTcpClient(server, port, SRS_EDGE_TOKEN_TRAVERSE_TMMS); + SrsTcpClient* transport = new SrsTcpClient(server, port, SRS_EDGE_TOKEN_TRAVERSE_TIMEOUT); SrsAutoFree(SrsTcpClient, transport); if ((err = transport->connect()) != srs_success) { diff --git a/trunk/src/app/srs_app_source.cpp b/trunk/src/app/srs_app_source.cpp index 73483efbc..def57ae5f 100755 --- a/trunk/src/app/srs_app_source.cpp +++ b/trunk/src/app/srs_app_source.cpp @@ -2515,6 +2515,7 @@ srs_error_t SrsSource::on_edge_start_publish() return publish_edge->on_client_publish(); } +// TODO: FIXME: Use edge strategy pattern. srs_error_t SrsSource::on_edge_proxy_publish(SrsCommonMessage* msg) { return publish_edge->on_proxy_publish(msg); diff --git a/trunk/src/kernel/srs_kernel_consts.hpp b/trunk/src/kernel/srs_kernel_consts.hpp index 4ffa78f49..4f22a581e 100644 --- a/trunk/src/kernel/srs_kernel_consts.hpp +++ b/trunk/src/kernel/srs_kernel_consts.hpp @@ -406,7 +406,7 @@ #define SRS_CONSTS_KAFKA_DEFAULT_PORT 9092 // the common io timeout, for both recv and send. -#define SRS_CONSTS_KAFKA_TMMS (30*1000) +#define SRS_CONSTS_KAFKA_TIMEOUT (30 * SRS_UTIME_MILLISECONDS) #endif diff --git a/trunk/src/service/srs_service_http_client.cpp b/trunk/src/service/srs_service_http_client.cpp index b90f116a4..e2c2163cc 100644 --- a/trunk/src/service/srs_service_http_client.cpp +++ b/trunk/src/service/srs_service_http_client.cpp @@ -220,7 +220,7 @@ srs_error_t SrsHttpClient::connect() return err; } - transport = new SrsTcpClient(host, port, timeout); + transport = new SrsTcpClient(host, port, srs_utime_t(timeout * SRS_UTIME_MILLISECONDS)); if ((err = transport->connect()) != srs_success) { disconnect(); return srs_error_wrap(err, "http: tcp connect %s:%d to=%d", host.c_str(), port, (int)timeout); diff --git a/trunk/src/service/srs_service_rtmp_conn.cpp b/trunk/src/service/srs_service_rtmp_conn.cpp index d1046db6d..4561c0a5d 100644 --- a/trunk/src/service/srs_service_rtmp_conn.cpp +++ b/trunk/src/service/srs_service_rtmp_conn.cpp @@ -65,7 +65,7 @@ srs_error_t SrsBasicRtmpClient::connect() close(); - transport = new SrsTcpClient(req->host, req->port, connect_timeout); + transport = new SrsTcpClient(req->host, req->port, srs_utime_t(connect_timeout * SRS_UTIME_MILLISECONDS)); client = new SrsRtmpClient(transport); kbps->set_io(transport, transport); diff --git a/trunk/src/service/srs_service_st.cpp b/trunk/src/service/srs_service_st.cpp index 395c44e41..9497a6f5e 100644 --- a/trunk/src/service/srs_service_st.cpp +++ b/trunk/src/service/srs_service_st.cpp @@ -104,12 +104,11 @@ srs_thread_t srs_thread_self() return (srs_thread_t)st_thread_self(); } -// TODO: FXIME: Refine tm in time unit. -srs_error_t srs_socket_connect(string server, int port, int64_t tm, srs_netfd_t* pstfd) +srs_error_t srs_socket_connect(string server, int port, srs_utime_t tm, srs_netfd_t* pstfd) { st_utime_t timeout = ST_UTIME_NO_TIMEOUT; if (tm != SRS_UTIME_NO_TIMEOUT) { - timeout = (st_utime_t)(tm * 1000); + timeout = tm; } *pstfd = NULL; @@ -416,7 +415,7 @@ srs_error_t SrsStSocket::writev(const iovec *iov, int iov_size, ssize_t* nwrite) return err; } -SrsTcpClient::SrsTcpClient(string h, int p, int64_t tm) +SrsTcpClient::SrsTcpClient(string h, int p, srs_utime_t tm) { stfd = NULL; io = new SrsStSocket(); @@ -441,7 +440,7 @@ srs_error_t SrsTcpClient::connect() srs_assert(stfd == NULL); if ((err = srs_socket_connect(host, port, timeout, &stfd)) != srs_success) { - return srs_error_wrap(err, "tcp: connect %s:%d to=%d", host.c_str(), port, (int)timeout); + return srs_error_wrap(err, "tcp: connect %s:%d to=%dms", host.c_str(), port, srsu2msi(timeout)); } if ((err = io->initialize(stfd)) != srs_success) { diff --git a/trunk/src/service/srs_service_st.hpp b/trunk/src/service/srs_service_st.hpp index f5eb9cffb..419117245 100644 --- a/trunk/src/service/srs_service_st.hpp +++ b/trunk/src/service/srs_service_st.hpp @@ -53,8 +53,8 @@ extern void srs_socket_reuse_addr(int fd); extern srs_thread_t srs_thread_self(); // client open socket and connect to server. -// @param tm The timeout in ms. -extern srs_error_t srs_socket_connect(std::string server, int port, int64_t tm, srs_netfd_t* pstfd); +// @param tm The timeout in srs_utime_t. +extern srs_error_t srs_socket_connect(std::string server, int port, srs_utime_t tm, srs_netfd_t* pstfd); // Wrap for coroutine. extern srs_cond_t srs_cond_new(); @@ -149,7 +149,7 @@ public: * The client to connect to server over TCP. * User must never reuse the client when close it. * Usage: - * SrsTcpClient client("127.0.0.1", 1935,9000); + * SrsTcpClient client("127.0.0.1", 1935, 9 * SRS_UTIME_SECONDS); * client.connect(); * client.write("Hello world!", 12, NULL); * client.read(buf, 4096, NULL); @@ -163,16 +163,16 @@ private: private: std::string host; int port; - // The timeout in ms. - int64_t timeout; + // The timeout in srs_utime_t. + srs_utime_t timeout; public: /** * Constructor. * @param h the ip or hostname of server. * @param p the port to connect to. - * @param tm the timeout in ms. + * @param tm the timeout in srs_utime_t. */ - SrsTcpClient(std::string h, int p, int64_t tm); + SrsTcpClient(std::string h, int p, srs_utime_t tm); virtual ~SrsTcpClient(); public: /**