srs-librtmp support hijack io apis for st-load. 2.0.42.

@@ -485,6 +485,7 @@ Supported operating systems and hardware:
 * 2013-10-17, Created.<br/>
 
 ## History
+* v2.0, 2014-12-02, srs-librtmp support hijack io apis for st-load. 2.0.42.
 * v2.0, 2014-12-01, for [#237](https://github.com/winlinvip/simple-rtmp-server/issues/237), refine syscall for recv, supports 1.5k clients. 2.0.41.
 * v2.0, 2014-11-30, add qtcreate project file trunk/src/qt/srs/srs-qt.pro. 2.0.39.
 * v2.0, 2014-11-29, fix [#235](https://github.com/winlinvip/simple-rtmp-server/issues/235), refine handshake, replace union with template method. 2.0.38.

@@ -31,7 +31,7 @@ CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
 // current release version
 #define VERSION_MAJOR       2
 #define VERSION_MINOR       0
-#define VERSION_REVISION    41
+#define VERSION_REVISION    42
 // server info.
 #define RTMP_SIG_SRS_KEY "SRS"
 #define RTMP_SIG_SRS_ROLE "origin/edge server"

@@ -43,188 +43,305 @@ CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
     #define ST_UTIME_NO_TIMEOUT -1
 #endif
 
+// when io not hijacked, use simple socket, the block sync stream.
+#ifndef SRS_HIJACK_IO
+    struct SrsBlockSyncSocket
+    {
+        SOCKET fd;
+        int64_t recv_timeout;
+        int64_t send_timeout;
+        int64_t recv_bytes;
+        int64_t send_bytes;
+        
+        SrsBlockSyncSocket() {
+            send_timeout = recv_timeout = ST_UTIME_NO_TIMEOUT;
+            recv_bytes = send_bytes = 0;
+            
+            SOCKET_RESET(fd);
+            SOCKET_SETUP();
+        }
+        
+        virtual ~SrsBlockSyncSocket() {
+            SOCKET_CLOSE(fd);
+            SOCKET_CLEANUP();
+        }
+    };
+    srs_hijack_io_t srs_hijack_io_create()
+    {
+        SrsBlockSyncSocket* skt = new SrsBlockSyncSocket();
+        return skt;
+    }
+    void srs_hijack_io_destroy(srs_hijack_io_t ctx)
+    {
+        SrsBlockSyncSocket* skt = (SrsBlockSyncSocket*)ctx;
+        srs_freep(skt);
+    }
+    int srs_hijack_io_create_socket(srs_hijack_io_t ctx)
+    {
+        SrsBlockSyncSocket* skt = (SrsBlockSyncSocket*)ctx;
+        
+        skt->fd = ::socket(AF_INET, SOCK_STREAM, 0);
+        if (!SOCKET_VALID(skt->fd)) {
+            return ERROR_SOCKET_CREATE;
+        }
+    
+        return ERROR_SUCCESS;
+    }
+    int srs_hijack_io_connect(srs_hijack_io_t ctx, const char* server_ip, int port)
+    {
+        SrsBlockSyncSocket* skt = (SrsBlockSyncSocket*)ctx;
+        
+        sockaddr_in addr;
+        addr.sin_family = AF_INET;
+        addr.sin_port = htons(port);
+        addr.sin_addr.s_addr = inet_addr(server_ip);
+        
+        if(::connect(skt->fd, (const struct sockaddr*)&addr, sizeof(sockaddr_in)) < 0){
+            return ERROR_SOCKET_CONNECT;
+        }
+        
+        return ERROR_SUCCESS;
+    }
+    int srs_hijack_io_read(srs_hijack_io_t ctx, void* buf, size_t size, ssize_t* nread)
+    {
+        SrsBlockSyncSocket* skt = (SrsBlockSyncSocket*)ctx;
+        
+        int ret = ERROR_SUCCESS;
+        
+        ssize_t nb_read = ::recv(skt->fd, (char*)buf, size, 0);
+        
+        if (nread) {
+            *nread = nb_read;
+        }
+        
+        // On success a non-negative integer indicating the number of bytes actually read is returned 
+        // (a value of 0 means the network connection is closed or end of file is reached).
+        if (nb_read <= 0) {
+            if (nb_read < 0 && SOCKET_ERRNO() == SOCKET_ETIME) {
+                return ERROR_SOCKET_TIMEOUT;
+            }
+            
+            if (nb_read == 0) {
+                errno = SOCKET_ECONNRESET;
+            }
+            
+            return ERROR_SOCKET_READ;
+        }
+        
+        skt->recv_bytes += nb_read;
+        
+        return ret;
+    }
+    void srs_hijack_io_set_recv_timeout(srs_hijack_io_t ctx, int64_t timeout_us)
+    {
+        SrsBlockSyncSocket* skt = (SrsBlockSyncSocket*)ctx;
+        skt->recv_timeout = timeout_us;
+    }
+    int64_t srs_hijack_io_get_recv_timeout(srs_hijack_io_t ctx)
+    {
+        SrsBlockSyncSocket* skt = (SrsBlockSyncSocket*)ctx;
+        return skt->recv_timeout;
+    }
+    int64_t srs_hijack_io_get_recv_bytes(srs_hijack_io_t ctx)
+    {
+        SrsBlockSyncSocket* skt = (SrsBlockSyncSocket*)ctx;
+        return skt->recv_bytes;
+    }
+    void srs_hijack_io_set_send_timeout(srs_hijack_io_t ctx, int64_t timeout_us)
+    {
+        SrsBlockSyncSocket* skt = (SrsBlockSyncSocket*)ctx;
+        skt->send_timeout = timeout_us;
+    }
+    int64_t srs_hijack_io_get_send_timeout(srs_hijack_io_t ctx)
+    {
+        SrsBlockSyncSocket* skt = (SrsBlockSyncSocket*)ctx;
+        return skt->send_timeout;
+    }
+    int64_t srs_hijack_io_get_send_bytes(srs_hijack_io_t ctx)
+    {
+        SrsBlockSyncSocket* skt = (SrsBlockSyncSocket*)ctx;
+        return skt->send_bytes;
+    }
+    int srs_hijack_io_writev(srs_hijack_io_t ctx, const iovec *iov, int iov_size, ssize_t* nwrite)
+    {
+        SrsBlockSyncSocket* skt = (SrsBlockSyncSocket*)ctx;
+        
+        int ret = ERROR_SUCCESS;
+        
+        ssize_t nb_write = ::writev(skt->fd, iov, iov_size);
+        
+        if (nwrite) {
+            *nwrite = nb_write;
+        }
+        
+        // On  success,  the  readv()  function  returns the number of bytes read; 
+        // the writev() function returns the number of bytes written.  On error, -1 is
+        // returned, and errno is set appropriately.
+        if (nb_write <= 0) {
+            // @see https://github.com/winlinvip/simple-rtmp-server/issues/200
+            if (nb_write < 0 && SOCKET_ERRNO() == SOCKET_ETIME) {
+                return ERROR_SOCKET_TIMEOUT;
+            }
+            
+            return ERROR_SOCKET_WRITE;
+        }
+        
+        skt->send_bytes += nb_write;
+        
+        return ret;
+    }
+    bool srs_hijack_io_is_never_timeout(srs_hijack_io_t ctx, int64_t timeout_us)
+    {
+        return timeout_us == (int64_t)ST_UTIME_NO_TIMEOUT;
+    }
+    int srs_hijack_io_read_fully(srs_hijack_io_t ctx, void* buf, size_t size, ssize_t* nread)
+    {
+        SrsBlockSyncSocket* skt = (SrsBlockSyncSocket*)ctx;
+        
+        int ret = ERROR_SUCCESS;
+        
+        size_t left = size;
+        ssize_t nb_read = 0;
+        
+        while (left > 0) {
+            char* this_buf = (char*)buf + nb_read;
+            ssize_t this_nread;
+            
+            if ((ret = srs_hijack_io_read(ctx, this_buf, left, &this_nread)) != ERROR_SUCCESS) {
+                return ret;
+            }
+            
+            nb_read += this_nread;
+            left -= (size_t)this_nread;
+        }
+        
+        if (nread) {
+            *nread = nb_read;
+        }
+        skt->recv_bytes += nb_read;
+        
+        return ret;
+    }
+    int srs_hijack_io_write(srs_hijack_io_t ctx, void* buf, size_t size, ssize_t* nwrite)
+    {
+        SrsBlockSyncSocket* skt = (SrsBlockSyncSocket*)ctx;
+        
+        int ret = ERROR_SUCCESS;
+        
+        ssize_t nb_write = ::send(skt->fd, (char*)buf, size, 0);
+        
+        if (nwrite) {
+            *nwrite = nb_write;
+        }
+        
+        if (nb_write <= 0) {
+            // @see https://github.com/winlinvip/simple-rtmp-server/issues/200
+            if (nb_write < 0 && SOCKET_ERRNO() == SOCKET_ETIME) {
+                return ERROR_SOCKET_TIMEOUT;
+            }
+            
+            return ERROR_SOCKET_WRITE;
+        }
+        
+        skt->send_bytes += nb_write;
+        
+        return ret;
+    }
+#endif
+
 SimpleSocketStream::SimpleSocketStream()
 {
-    SOCKET_RESET(fd);
-    send_timeout = recv_timeout = ST_UTIME_NO_TIMEOUT;
-    recv_bytes = send_bytes = 0;
-    SOCKET_SETUP();
+    io = srs_hijack_io_create();
 }
 
 SimpleSocketStream::~SimpleSocketStream()
 {
-    SOCKET_CLOSE(fd);
-    SOCKET_CLEANUP();
+    if (io) {
+        srs_hijack_io_destroy(io);
+        io = NULL;
+    }
 }
 
 int SimpleSocketStream::create_socket()
 {
-    fd = ::socket(AF_INET, SOCK_STREAM, 0);
-    if (!SOCKET_VALID(fd)) {
-        return ERROR_SOCKET_CREATE;
-    }
-
-    return ERROR_SUCCESS;
+    srs_assert(io);
+    return srs_hijack_io_create_socket(io);
 }
 
 int SimpleSocketStream::connect(const char* server_ip, int port)
 {
-    sockaddr_in addr;
-    addr.sin_family = AF_INET;
-    addr.sin_port = htons(port);
-    addr.sin_addr.s_addr = inet_addr(server_ip);
-    
-    if(::connect(fd, (const struct sockaddr*)&addr, sizeof(sockaddr_in)) < 0){
-        return ERROR_SOCKET_CONNECT;
-    }
-
-    return ERROR_SUCCESS;
+    srs_assert(io);
+    return srs_hijack_io_connect(io, server_ip, port);
 }
 
 // ISrsBufferReader
 int SimpleSocketStream::read(void* buf, size_t size, ssize_t* nread)
 {
-    int ret = ERROR_SUCCESS;
-    
-    ssize_t nb_read = ::recv(fd, (char*)buf, size, 0);
-    
-    if (nread) {
-        *nread = nb_read;
-    }
-    
-    // On success a non-negative integer indicating the number of bytes actually read is returned 
-    // (a value of 0 means the network connection is closed or end of file is reached).
-    if (nb_read <= 0) {
-        if (nb_read < 0 && SOCKET_ERRNO() == SOCKET_ETIME) {
-            return ERROR_SOCKET_TIMEOUT;
-        }
-        
-        if (nb_read == 0) {
-            errno = SOCKET_ECONNRESET;
-        }
-        
-        return ERROR_SOCKET_READ;
-    }
-    
-    recv_bytes += nb_read;
-    
-    return ret;
+    srs_assert(io);
+    return srs_hijack_io_read(io, buf, size, nread);
 }
 
 // ISrsProtocolReader
 void SimpleSocketStream::set_recv_timeout(int64_t timeout_us)
 {
-    recv_timeout = timeout_us;
+    srs_assert(io);
+    srs_hijack_io_set_recv_timeout(io, timeout_us);
 }
 
 int64_t SimpleSocketStream::get_recv_timeout()
 {
-    return recv_timeout;
+    srs_assert(io);
+    return srs_hijack_io_get_recv_timeout(io);
 }
 
 int64_t SimpleSocketStream::get_recv_bytes()
 {
-    return recv_bytes;
+    srs_assert(io);
+    return srs_hijack_io_get_recv_bytes(io);
 }
 
 // ISrsProtocolWriter
 void SimpleSocketStream::set_send_timeout(int64_t timeout_us)
 {
-    send_timeout = timeout_us;
+    srs_assert(io);
+    srs_hijack_io_set_send_timeout(io, timeout_us);
 }
 
 int64_t SimpleSocketStream::get_send_timeout()
 {
-    return send_timeout;
+    srs_assert(io);
+    return srs_hijack_io_get_send_timeout(io);
 }
 
 int64_t SimpleSocketStream::get_send_bytes()
 {
-    return send_bytes;
+    srs_assert(io);
+    return srs_hijack_io_get_send_bytes(io);
 }
 
 int SimpleSocketStream::writev(const iovec *iov, int iov_size, ssize_t* nwrite)
 {
-    int ret = ERROR_SUCCESS;
-    
-    ssize_t nb_write = ::writev(fd, iov, iov_size);
-    
-    if (nwrite) {
-        *nwrite = nb_write;
-    }
-    
-    // On  success,  the  readv()  function  returns the number of bytes read; 
-    // the writev() function returns the number of bytes written.  On error, -1 is
-    // returned, and errno is set appropriately.
-    if (nb_write <= 0) {
-        // @see https://github.com/winlinvip/simple-rtmp-server/issues/200
-        if (nb_write < 0 && SOCKET_ERRNO() == SOCKET_ETIME) {
-            return ERROR_SOCKET_TIMEOUT;
-        }
-        
-        return ERROR_SOCKET_WRITE;
-    }
-    
-    send_bytes += nb_write;
-    
-    return ret;
+    srs_assert(io);
+    return srs_hijack_io_writev(io, iov, iov_size, nwrite);
 }
 
 // ISrsProtocolReaderWriter
 bool SimpleSocketStream::is_never_timeout(int64_t timeout_us)
 {
-    return timeout_us == (int64_t)ST_UTIME_NO_TIMEOUT;
+    srs_assert(io);
+    return srs_hijack_io_is_never_timeout(io, timeout_us);
 }
 
 int SimpleSocketStream::read_fully(void* buf, size_t size, ssize_t* nread)
 {
-    int ret = ERROR_SUCCESS;
-    
-    size_t left = size;
-    ssize_t nb_read = 0;
-    
-    while (left > 0) {
-        char* this_buf = (char*)buf + nb_read;
-        ssize_t this_nread;
-        
-        if ((ret = this->read(this_buf, left, &this_nread)) != ERROR_SUCCESS) {
-            return ret;
-        }
-        
-        nb_read += this_nread;
-        left -= (size_t)this_nread;
-    }
-    
-    if (nread) {
-        *nread = nb_read;
-    }
-    recv_bytes += nb_read;
-    
-    return ret;
+    srs_assert(io);
+    return srs_hijack_io_read_fully(io, buf, size, nread);
 }
 
 int SimpleSocketStream::write(void* buf, size_t size, ssize_t* nwrite)
 {
-    int ret = ERROR_SUCCESS;
-    
-    ssize_t nb_write = ::send(fd, (char*)buf, size, 0);
-    
-    if (nwrite) {
-        *nwrite = nb_write;
-    }
-    
-    if (nb_write <= 0) {
-        // @see https://github.com/winlinvip/simple-rtmp-server/issues/200
-        if (nb_write < 0 && SOCKET_ERRNO() == SOCKET_ETIME) {
-            return ERROR_SOCKET_TIMEOUT;
-        }
-        
-        return ERROR_SOCKET_WRITE;
-    }
-    
-    send_bytes += nb_write;
-    
-    return ret;
+    srs_assert(io);
+    return srs_hijack_io_write(io, buf, size, nwrite);
 }
 
 

@@ -32,7 +32,7 @@ CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
 
 #include <srs_protocol_io.hpp>
 #include <srs_librtmp.hpp>
-    
+
 /**
 * simple socket stream,
 * use tcp socket, sync block mode, for client like srs-librtmp.
@@ -40,11 +40,7 @@ CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
 class SimpleSocketStream : public ISrsProtocolReaderWriter
 {
 private:
-    int64_t recv_timeout;
-    int64_t send_timeout;
-    int64_t recv_bytes;
-    int64_t send_bytes;
-    SOCKET fd;
+    srs_hijack_io_t io;
 public:
     SimpleSocketStream();
     virtual ~SimpleSocketStream();

@@ -49,7 +49,7 @@ using namespace std;
 #include <srs_lib_bandwidth.hpp>
 
 // if want to use your log, define the folowing macro.
-#ifndef SRS_RTMP_USER_DEFINED_LOG
+#ifndef SRS_HIJACK_LOG
     // kernel module.
     ISrsLog* _srs_log = new ISrsLog();
     ISrsThreadContext* _srs_context = new ISrsThreadContext();

@@ -930,6 +930,93 @@ extern const char* srs_human_format_time();
 #define srs_human_verbose(msg, ...) printf("[%s] ", srs_human_format_time());printf(msg, ##__VA_ARGS__);printf("\n")
 #define srs_human_raw(msg, ...) printf(msg, ##__VA_ARGS__)
 
+/*************************************************************
+**************************************************************
+* IO hijack, use your specified io functions.
+**************************************************************
+*************************************************************/
+// the void* will convert to your handler for io hijack.
+typedef void* srs_hijack_io_t;
+// define the following macro and functions in your module to hijack the io.
+// the example @see https://github.com/winlinvip/st-load
+// which use librtmp but use its own io(use st also).
+#ifdef SRS_HIJACK_IO
+    /**
+    * create hijack.
+    * @return NULL for error; otherwise, ok.
+    */
+    extern srs_hijack_io_t srs_hijack_io_create();
+    /**
+    * destroy the context, user must close the socket.
+    */
+    extern void srs_hijack_io_destroy(srs_hijack_io_t ctx);
+    /**
+    * create socket, not connect yet.
+    * @return 0, success; otherswise, failed.
+    */
+    extern int srs_hijack_io_create_socket(srs_hijack_io_t ctx);
+    /**
+    * connect socket at server_ip:port.
+    * @return 0, success; otherswise, failed.
+    */
+    extern int srs_hijack_io_connect(srs_hijack_io_t ctx, const char* server_ip, int port);
+    /**
+    * read from socket.
+    * @return 0, success; otherswise, failed.
+    */
+    extern int srs_hijack_io_read(srs_hijack_io_t ctx, void* buf, size_t size, ssize_t* nread);
+    /**
+    * set the socket recv timeout.
+    * @return 0, success; otherswise, failed.
+    */
+    extern void srs_hijack_io_set_recv_timeout(srs_hijack_io_t ctx, int64_t timeout_us);
+    /**
+    * get the socket recv timeout.
+    * @return 0, success; otherswise, failed.
+    */
+    extern int64_t srs_hijack_io_get_recv_timeout(srs_hijack_io_t ctx);
+    /**
+    * get the socket recv bytes.
+    * @return 0, success; otherswise, failed.
+    */
+    extern int64_t srs_hijack_io_get_recv_bytes(srs_hijack_io_t ctx);
+    /**
+    * set the socket send timeout.
+    * @return 0, success; otherswise, failed.
+    */
+    extern void srs_hijack_io_set_send_timeout(srs_hijack_io_t ctx, int64_t timeout_us);
+    /**
+    * get the socket send timeout.
+    * @return 0, success; otherswise, failed.
+    */
+    extern int64_t srs_hijack_io_get_send_timeout(srs_hijack_io_t ctx);
+    /**
+    * get the socket send bytes.
+    * @return 0, success; otherswise, failed.
+    */
+    extern int64_t srs_hijack_io_get_send_bytes(srs_hijack_io_t ctx);
+    /**
+    * writev of socket.
+    * @return 0, success; otherswise, failed.
+    */
+    extern int srs_hijack_io_writev(srs_hijack_io_t ctx, const iovec *iov, int iov_size, ssize_t* nwrite);
+    /**
+    * whether the timeout is never timeout.
+    * @return 0, success; otherswise, failed.
+    */
+    extern bool srs_hijack_io_is_never_timeout(srs_hijack_io_t ctx, int64_t timeout_us);
+    /**
+    * read fully, fill the buf exactly size bytes.
+    * @return 0, success; otherswise, failed.
+    */
+    extern int srs_hijack_io_read_fully(srs_hijack_io_t ctx, void* buf, size_t size, ssize_t* nread);
+    /**
+    * write bytes to socket.
+    * @return 0, success; otherswise, failed.
+    */
+    extern int srs_hijack_io_write(srs_hijack_io_t ctx, void* buf, size_t size, ssize_t* nwrite);
+#endif
+
 #ifdef __cplusplus
 }
 #endif