Merge branch '3.0release' into develop

README.md

@@ -6,7 +6,7 @@
 [<img width="52" alt="Skype" src="https://cloud.githubusercontent.com/assets/2777660/24329166/3821a328-1230-11e7-844a-506a5d17dd3d.png">](https://github.com/ossrs/srs/wiki/v1_EN_Contact#skype-or-gitter)
 
 SRS/3.0，[OuXuli][release3]，是一个简单的流媒体直播集群，简单的快乐。<br/>
-SRS, a simple joy, created on 2013.10, is a simple live streaming cluster.
+SRS is a simple live streaming cluster, a simple joy.
 
 Download binaries from github.io: [Centos6-x86_64][centos0], [more...][more0]<br/>
 Download binaries from ossrs.net: [Centos6-x86_64][centos1], [more...][more1]<br/>

trunk/src/app/srs_app_hls.cpp

@@ -462,6 +462,9 @@ srs_error_t SrsHlsMuxer::segment_open()
     if ((err = current->writer->open(tmp_file)) != srs_success) {
         return srs_error_wrap(err, "open hls muxer");
     }
+
+    // reset the context for a new ts start.
+    context->reset();
     
     return err;
 }

trunk/src/app/srs_app_hls.hpp

@@ -157,7 +157,8 @@ private:
     // The key and iv.
     unsigned char key[16];
     unsigned char iv[16];
-    SrsFileWriter *writer;
+    // The underlayer file writer.
+    SrsFileWriter* writer;
 private:
     int _sequence_no;
     srs_utime_t max_td;

trunk/src/app/srs_app_listener.cpp

@@ -46,9 +46,6 @@ using namespace std;
 // sleep in srs_utime_t for udp recv packet.
 #define SrsUdpPacketRecvCycleInterval 0
 
-// nginx also set to 512
-#define SERVER_LISTEN_BACKLOG 512
-
 ISrsUdpHandler::ISrsUdpHandler()
 {
 }
@@ -75,9 +72,7 @@ SrsUdpListener::SrsUdpListener(ISrsUdpHandler* h, string i, int p)
     handler = h;
     ip = i;
     port = p;
-    
+    lfd = NULL;
-    _fd = -1;
-    _stfd = NULL;
     
     nb_buf = SRS_UDP_MAX_PACKET_SIZE;
     buf = new char[nb_buf];
@@ -87,60 +82,27 @@ SrsUdpListener::SrsUdpListener(ISrsUdpHandler* h, string i, int p)
 
 SrsUdpListener::~SrsUdpListener()
 {
-    // close the stfd to trigger thread to interrupted.
-    srs_close_stfd(_stfd);
-    
     srs_freep(trd);
-    
+    srs_close_stfd(lfd);
-    // st does not close it sometimes,
-    // close it manually.
-    close(_fd);
-    
     srs_freepa(buf);
 }
 
 int SrsUdpListener::fd()
 {
-    return _fd;
+    return srs_netfd_fileno(lfd);
 }
 
 srs_netfd_t SrsUdpListener::stfd()
 {
-    return _stfd;
+    return lfd;
 }
 
 srs_error_t SrsUdpListener::listen()
 {
     srs_error_t err = srs_success;
-    
-    char sport[8];
-    snprintf(sport, sizeof(sport), "%d", port);
-    
-    addrinfo hints;
-    memset(&hints, 0, sizeof(hints));
-    hints.ai_family   = AF_UNSPEC;
-    hints.ai_socktype = SOCK_DGRAM;
-    hints.ai_flags    = AI_NUMERICHOST;
-    
-    addrinfo* r  = NULL;
-    SrsAutoFree(addrinfo, r);
-    if(getaddrinfo(ip.c_str(), sport, (const addrinfo*)&hints, &r)) {
-        return srs_error_new(ERROR_SYSTEM_IP_INVALID, "get address info");
-    }
-    
-    if ((_fd = socket(r->ai_family, r->ai_socktype, r->ai_protocol)) == -1) {
-        return srs_error_new(ERROR_SOCKET_CREATE, "create socket. ip=%s, port=%d", ip.c_str(), port);
-    }
 
-    srs_fd_close_exec(_fd);
+    if ((err = srs_udp_listen(ip, port, &lfd)) != srs_success) {
-    srs_socket_reuse_addr(_fd);
+        return srs_error_wrap(err, "listen %s:%d", ip.c_str(), port);
-    
-    if (bind(_fd, r->ai_addr, r->ai_addrlen) == -1) {
-        return srs_error_new(ERROR_SOCKET_BIND, "bind socket. ep=%s:%d", ip.c_str(), port);;
-    }
-    
-    if ((_stfd = srs_netfd_open_socket(_fd)) == NULL){
-        return srs_error_new(ERROR_ST_OPEN_SOCKET, "st open socket");
     }
     
     srs_freep(trd);
@@ -160,12 +122,11 @@ srs_error_t SrsUdpListener::cycle()
         if ((err = trd->pull()) != srs_success) {
             return srs_error_wrap(err, "udp listener");
         }
-        
+
+        int nread = 0;
         sockaddr_storage from;
         int nb_from = sizeof(from);
-        int nread = 0;
+        if ((nread = srs_recvfrom(lfd, buf, nb_buf, (sockaddr*)&from, &nb_from, SRS_UTIME_NO_TIMEOUT)) <= 0) {
-        
-        if ((nread = srs_recvfrom(_stfd, buf, nb_buf, (sockaddr*)&from, &nb_from, SRS_UTIME_NO_TIMEOUT)) <= 0) {
             return srs_error_new(ERROR_SOCKET_READ, "udp read, nread=%d", nread);
         }
         
@@ -186,9 +147,8 @@ SrsTcpListener::SrsTcpListener(ISrsTcpHandler* h, string i, int p)
     handler = h;
     ip = i;
     port = p;
-    
+
-    _fd = -1;
+    lfd = NULL;
-    _stfd = NULL;
     
     trd = new SrsDummyCoroutine();
 }
@@ -196,60 +156,20 @@ SrsTcpListener::SrsTcpListener(ISrsTcpHandler* h, string i, int p)
 SrsTcpListener::~SrsTcpListener()
 {
     srs_freep(trd);
-    
+    srs_close_stfd(lfd);
-    srs_close_stfd(_stfd);
 }
 
 int SrsTcpListener::fd()
 {
-    return _fd;
+    return srs_netfd_fileno(lfd);;
 }
 
 srs_error_t SrsTcpListener::listen()
 {
     srs_error_t err = srs_success;
-    
-    char sport[8];
-    snprintf(sport, sizeof(sport), "%d", port);
-    
-    addrinfo hints;
-    memset(&hints, 0, sizeof(hints));
-    hints.ai_family   = AF_UNSPEC;
-    hints.ai_socktype = SOCK_STREAM;
-    hints.ai_flags    = AI_NUMERICHOST;
-    
-    addrinfo* r = NULL;
-    SrsAutoFree(addrinfo, r);
-    if(getaddrinfo(ip.c_str(), sport, (const addrinfo*)&hints, &r)) {
-        return srs_error_new(ERROR_SYSTEM_IP_INVALID, "get address info");
-    }
-    
-    if ((_fd = socket(r->ai_family, r->ai_socktype, r->ai_protocol)) == -1) {
-        return srs_error_new(ERROR_SOCKET_CREATE, "create socket. ip=%s, port=%d", ip.c_str(), port);
-    }
-    
-    // Detect alive for TCP connection.
-    // @see https://github.com/ossrs/srs/issues/1044
-#ifdef SO_KEEPALIVE
-    int tcp_keepalive = 1;
-    if (setsockopt(_fd, SOL_SOCKET, SO_KEEPALIVE, &tcp_keepalive, sizeof(int)) == -1) {
-        return srs_error_new(ERROR_SOCKET_SETKEEPALIVE, "setsockopt SO_KEEPALIVE[%d]error. port=%d", tcp_keepalive, port);
-    }
-#endif
 
-    srs_fd_close_exec(_fd);
+    if ((err = srs_tcp_listen(ip, port, &lfd)) != srs_success) {
-    srs_socket_reuse_addr(_fd);
+        return srs_error_wrap(err, "listen at %s:%d", ip.c_str(), port);
-    
-    if (bind(_fd, r->ai_addr, r->ai_addrlen) == -1) {
-        return srs_error_new(ERROR_SOCKET_BIND, "bind socket. ep=%s:%d", ip.c_str(), port);;
-    }
-
-    if (::listen(_fd, SERVER_LISTEN_BACKLOG) == -1) {
-        return srs_error_new(ERROR_SOCKET_LISTEN, "listen socket");
-    }
-    
-    if ((_stfd = srs_netfd_open_socket(_fd)) == NULL){
-        return srs_error_new(ERROR_ST_OPEN_SOCKET, "st open socket");
     }
     
     srs_freep(trd);
@@ -270,16 +190,17 @@ srs_error_t SrsTcpListener::cycle()
             return srs_error_wrap(err, "tcp listener");
         }
         
-        srs_netfd_t cstfd = srs_accept(_stfd, NULL, NULL, SRS_UTIME_NO_TIMEOUT);
+        srs_netfd_t fd = srs_accept(lfd, NULL, NULL, SRS_UTIME_NO_TIMEOUT);
-        if(cstfd == NULL){
+        if(fd == NULL){
-            return srs_error_new(ERROR_SOCKET_CREATE, "accept failed");
+            return srs_error_new(ERROR_SOCKET_ACCEPT, "accept at fd=%d", srs_netfd_fileno(lfd));
         }
         
-        int cfd = srs_netfd_fileno(cstfd);
+	    if ((err = srs_fd_closeexec(srs_netfd_fileno(fd))) != srs_success) {
-        srs_fd_close_exec(cfd);
+	        return srs_error_wrap(err, "set closeexec");
+	    }
         
-        if ((err = handler->on_tcp_client(cstfd)) != srs_success) {
+        if ((err = handler->on_tcp_client(fd)) != srs_success) {
-            return srs_error_wrap(err, "handle fd=%d", cfd);
+            return srs_error_wrap(err, "handle fd=%d", srs_netfd_fileno(fd));
         }
     }
     

trunk/src/app/srs_app_listener.hpp

@@ -69,8 +69,7 @@ public:
 class SrsUdpListener : public ISrsCoroutineHandler
 {
 private:
-    int _fd;
+    srs_netfd_t lfd;
-    srs_netfd_t _stfd;
     SrsCoroutine* trd;
 private:
     char* buf;
@@ -96,8 +95,7 @@ public:
 class SrsTcpListener : public ISrsCoroutineHandler
 {
 private:
-    int _fd;
+    srs_netfd_t lfd;
-    srs_netfd_t _stfd;
     SrsCoroutine* trd;
 private:
     ISrsTcpHandler* handler;

trunk/src/kernel/srs_kernel_error.hpp

@@ -115,6 +115,9 @@
 #define ERROR_SOCKET_NO_NODELAY             1076
 #define ERROR_SOCKET_SNDBUF                 1077
 #define ERROR_THREAD_STARTED                1078
+#define ERROR_SOCKET_SETREUSEADDR           1079
+#define ERROR_SOCKET_SETCLOSEEXEC           1080
+#define ERROR_SOCKET_ACCEPT                 1081
 
 ///////////////////////////////////////////////////////
 // RTMP protocol error.

trunk/src/service/srs_service_st.cpp

@@ -35,6 +35,9 @@ using namespace std;
 #include <srs_service_utility.hpp>
 #include <srs_kernel_utility.hpp>
 
+// nginx also set to 512
+#define SERVER_LISTEN_BACKLOG 512
+
 #ifdef __linux__
 #include <sys/epoll.h>
 
@@ -86,17 +89,37 @@ void srs_close_stfd(srs_netfd_t& stfd)
     }
 }
 
-void srs_fd_close_exec(int fd)
+srs_error_t srs_fd_closeexec(int fd)
 {
     int flags = fcntl(fd, F_GETFD);
     flags |= FD_CLOEXEC;
-    fcntl(fd, F_SETFD, flags);
+    if (fcntl(fd, F_SETFD, flags) == -1) {
+        return srs_error_new(ERROR_SOCKET_SETCLOSEEXEC, "FD_CLOEXEC fd=%v", fd);
+    }
+
+    return srs_success;
 }
 
-void srs_socket_reuse_addr(int fd)
+srs_error_t srs_fd_reuseaddr(int fd)
 {
     int v = 1;
-    setsockopt(fd, SOL_SOCKET, SO_REUSEADDR, &v, sizeof(int));
+    if (setsockopt(fd, SOL_SOCKET, SO_REUSEADDR, &v, sizeof(int)) == -1) {
+        return srs_error_new(ERROR_SOCKET_SETREUSEADDR, "SO_REUSEADDR fd=%v", fd);
+    }
+
+	return srs_success;
+}
+
+srs_error_t srs_fd_keepalive(int fd)
+{
+#ifdef SO_KEEPALIVE
+    int v = 1;
+    if (setsockopt(fd, SOL_SOCKET, SO_KEEPALIVE, &v, sizeof(int)) == -1) {
+        return srs_error_new(ERROR_SOCKET_SETKEEPALIVE, "SO_KEEPALIVE fd=%d", fd);
+    }
+#endif
+
+	return srs_success;
 }
 
 srs_thread_t srs_thread_self()
@@ -104,7 +127,7 @@ srs_thread_t srs_thread_self()
     return (srs_thread_t)st_thread_self();
 }
 
-srs_error_t srs_socket_connect(string server, int port, srs_utime_t tm, srs_netfd_t* pstfd)
+srs_error_t srs_tcp_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) {
@@ -149,6 +172,116 @@ srs_error_t srs_socket_connect(string server, int port, srs_utime_t tm, srs_netf
     return srs_success;
 }
 
+srs_error_t srs_tcp_listen(std::string ip, int port, srs_netfd_t* pfd)
+{
+	srs_error_t err = srs_success;
+
+    char sport[8];
+    snprintf(sport, sizeof(sport), "%d", port);
+
+    addrinfo hints;
+    memset(&hints, 0, sizeof(hints));
+    hints.ai_family   = AF_UNSPEC;
+    hints.ai_socktype = SOCK_STREAM;
+    hints.ai_flags    = AI_NUMERICHOST;
+
+    addrinfo* r = NULL;
+    SrsAutoFree(addrinfo, r);
+    if(getaddrinfo(ip.c_str(), sport, (const addrinfo*)&hints, &r)) {
+        return srs_error_new(ERROR_SYSTEM_IP_INVALID, "getaddrinfo hints=(%d,%d,%d)",
+            hints.ai_family, hints.ai_socktype, hints.ai_flags);
+    }
+
+    int fd = 0;
+    if ((fd = socket(r->ai_family, r->ai_socktype, r->ai_protocol)) == -1) {
+        return srs_error_new(ERROR_SOCKET_CREATE, "socket domain=%d, type=%d, protocol=%d",
+            r->ai_family, r->ai_socktype, r->ai_protocol);
+    }
+
+    // Detect alive for TCP connection.
+    // @see https://github.com/ossrs/srs/issues/1044
+    if ((err = srs_fd_keepalive(fd)) != srs_success) {
+        ::close(fd);
+        return srs_error_wrap(err, "set keepalive fd=%d", fd);
+    }
+
+    if ((err = srs_fd_closeexec(fd)) != srs_success) {
+        ::close(fd);
+        return srs_error_wrap(err, "set closeexec fd=%d", fd);
+    }
+
+    if ((err = srs_fd_reuseaddr(fd)) != srs_success) {
+        ::close(fd);
+        return srs_error_wrap(err, "set reuseaddr fd=%d", fd);
+    }
+
+    if (bind(fd, r->ai_addr, r->ai_addrlen) == -1) {
+        ::close(fd);
+        return srs_error_new(ERROR_SOCKET_BIND, "bind fd=%d", fd);
+    }
+
+    if (::listen(fd, SERVER_LISTEN_BACKLOG) == -1) {
+        ::close(fd);
+        return srs_error_new(ERROR_SOCKET_LISTEN, "listen fd=%d", fd);
+    }
+
+    if ((*pfd = srs_netfd_open_socket(fd)) == NULL){
+        ::close(fd);
+        return srs_error_new(ERROR_ST_OPEN_SOCKET, "st open fd=%d", fd);
+    }
+
+    return err;
+}
+
+srs_error_t srs_udp_listen(std::string ip, int port, srs_netfd_t* pfd)
+{
+	srs_error_t err = srs_success;
+
+    char sport[8];
+    snprintf(sport, sizeof(sport), "%d", port);
+
+    addrinfo hints;
+    memset(&hints, 0, sizeof(hints));
+    hints.ai_family   = AF_UNSPEC;
+    hints.ai_socktype = SOCK_DGRAM;
+    hints.ai_flags    = AI_NUMERICHOST;
+
+    addrinfo* r  = NULL;
+    SrsAutoFree(addrinfo, r);
+    if(getaddrinfo(ip.c_str(), sport, (const addrinfo*)&hints, &r)) {
+        return srs_error_new(ERROR_SYSTEM_IP_INVALID, "getaddrinfo hints=(%d,%d,%d)",
+            hints.ai_family, hints.ai_socktype, hints.ai_flags);
+    }
+
+	int fd = 0;
+    if ((fd = socket(r->ai_family, r->ai_socktype, r->ai_protocol)) == -1) {
+        return srs_error_new(ERROR_SOCKET_CREATE, "socket domain=%d, type=%d, protocol=%d",
+            r->ai_family, r->ai_socktype, r->ai_protocol);
+    }
+
+    if ((err = srs_fd_closeexec(fd)) != srs_success) {
+        ::close(fd);
+        return srs_error_wrap(err, "set closeexec fd=%d", fd);
+    }
+
+    if ((err = srs_fd_reuseaddr(fd)) != srs_success) {
+        ::close(fd);
+        return srs_error_wrap(err, "set reuseaddr fd=%d", fd);
+    }
+
+    if (bind(fd, r->ai_addr, r->ai_addrlen) == -1) {
+        ::close(fd);
+        return srs_error_new(ERROR_SOCKET_BIND, "bind fd=%d", fd);
+    }
+
+    if ((*pfd = srs_netfd_open_socket(fd)) == NULL){
+        ::close(fd);
+        return srs_error_new(ERROR_ST_OPEN_SOCKET, "st open fd=%d", fd);
+    }
+
+    return err;
+}
+
 srs_cond_t srs_cond_new()
 {
     return (srs_cond_t)st_cond_new();
@@ -439,7 +572,7 @@ srs_error_t SrsTcpClient::connect()
     close();
     
     srs_assert(stfd == NULL);
-    if ((err = srs_socket_connect(host, port, timeout, &stfd)) != srs_success) {
+    if ((err = srs_tcp_connect(host, port, timeout, &stfd)) != srs_success) {
         return srs_error_wrap(err, "tcp: connect %s:%d to=%dms", host.c_str(), port, srsu2msi(timeout));
     }
     

trunk/src/service/srs_service_st.hpp

@@ -44,17 +44,26 @@ extern srs_error_t srs_st_init();
 extern void srs_close_stfd(srs_netfd_t& stfd);
 
 // Set the FD_CLOEXEC of FD.
-extern void srs_fd_close_exec(int fd);
+extern srs_error_t srs_fd_closeexec(int fd);
 
-// Set the SO_REUSEADDR of socket.
+// Set the SO_REUSEADDR of fd.
-extern void srs_socket_reuse_addr(int fd);
+extern srs_error_t srs_fd_reuseaddr(int fd);
+
+// Set the SO_KEEPALIVE of fd.
+extern srs_error_t srs_fd_keepalive(int fd);
 
 // Get current coroutine/thread.
 extern srs_thread_t srs_thread_self();
 
-// client open socket and connect to server.
+// For client, to open socket and connect to server.
 // @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);
+extern srs_error_t srs_tcp_connect(std::string server, int port, srs_utime_t tm, srs_netfd_t* pstfd);
+
+// For server, listen at TCP endpoint.
+extern srs_error_t srs_tcp_listen(std::string ip, int port, srs_netfd_t* pfd);
+
+// For server, listen at UDP endpoint.
+extern srs_error_t srs_udp_listen(std::string ip, int port, srs_netfd_t* pfd);
 
 // Wrap for coroutine.
 extern srs_cond_t srs_cond_new();

trunk/src/utest/srs_utest_app.cpp

@@ -377,3 +377,9 @@ VOID TEST(AppFragmentTest, CheckDuration)
 	}
 }
 
+#define MOCK_LISTEN_PORT 11935
+
+VOID TEST(TCPServerTest, PingPong)
+{
+}
+