Refine SrsTcpClient timeout in time unit

trunk/src/app/srs_app_dash.cpp

@@ -172,6 +172,13 @@ 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);
@@ -182,9 +189,14 @@ srs_error_t SrsMpdWriter::initialize(SrsRequest* r)
     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);
 }

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:

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);

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) {

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);

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
 

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);

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);
     

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, srs_utime_t tm, srs_netfd_t* pstfd)
-srs_error_t srs_socket_connect(string server, int port, int64_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) {

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.
+// @param tm The timeout in srs_utime_t.
-extern srs_error_t srs_socket_connect(std::string server, int port, int64_t tm, srs_netfd_t* pstfd);
+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.
+    // The timeout in srs_utime_t.
-    int64_t timeout;
+    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:
     /**