proxy to srs ok

trunk/src/core/dlp_core.cpp

@@ -277,3 +277,159 @@ int dlp_socket_connect(string server, int port, st_utime_t timeout, st_netfd_t*
     return ret;
 }
 
+DlpStSocket::DlpStSocket(st_netfd_t client_stfd)
+{
+    stfd = client_stfd;
+    send_timeout = recv_timeout = ST_UTIME_NO_TIMEOUT;
+    recv_bytes = send_bytes = 0;
+}
+
+DlpStSocket::~DlpStSocket()
+{
+}
+
+bool DlpStSocket::is_never_timeout(int64_t timeout_us)
+{
+    return timeout_us == (int64_t)ST_UTIME_NO_TIMEOUT;
+}
+
+void DlpStSocket::set_recv_timeout(int64_t timeout_us)
+{
+    recv_timeout = timeout_us;
+}
+
+int64_t DlpStSocket::get_recv_timeout()
+{
+    return recv_timeout;
+}
+
+void DlpStSocket::set_send_timeout(int64_t timeout_us)
+{
+    send_timeout = timeout_us;
+}
+
+int64_t DlpStSocket::get_send_timeout()
+{
+    return send_timeout;
+}
+
+int64_t DlpStSocket::get_recv_bytes()
+{
+    return recv_bytes;
+}
+
+int64_t DlpStSocket::get_send_bytes()
+{
+    return send_bytes;
+}
+
+int DlpStSocket::read(void* buf, size_t size, ssize_t* nread)
+{
+    int ret = ERROR_SUCCESS;
+
+    ssize_t nb_read = st_read(stfd, buf, size, recv_timeout);
+    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).
+    // Otherwise, a value of -1 is returned and errno is set to indicate the error.
+    if (nb_read <= 0) {
+        // @see https://github.com/simple-rtmp-server/srs/issues/200
+        if (nb_read < 0 && errno == ETIME) {
+            return ERROR_SOCKET_TIMEOUT;
+        }
+
+        if (nb_read == 0) {
+            errno = ECONNRESET;
+        }
+
+        return ERROR_SOCKET_READ;
+    }
+
+    recv_bytes += nb_read;
+
+    return ret;
+}
+
+int DlpStSocket::read_fully(void* buf, size_t size, ssize_t* nread)
+{
+    int ret = ERROR_SUCCESS;
+
+    ssize_t nb_read = st_read_fully(stfd, buf, size, recv_timeout);
+    if (nread) {
+        *nread = nb_read;
+    }
+
+    // On success a non-negative integer indicating the number of bytes actually read is returned
+    // (a value less than nbyte means the network connection is closed or end of file is reached)
+    // Otherwise, a value of -1 is returned and errno is set to indicate the error.
+    if (nb_read != (ssize_t)size) {
+        // @see https://github.com/simple-rtmp-server/srs/issues/200
+        if (nb_read < 0 && errno == ETIME) {
+            return ERROR_SOCKET_TIMEOUT;
+        }
+
+        if (nb_read >= 0) {
+            errno = ECONNRESET;
+        }
+
+        return ERROR_SOCKET_READ_FULLY;
+    }
+
+    recv_bytes += nb_read;
+
+    return ret;
+}
+
+int DlpStSocket::write(void* buf, size_t size, ssize_t* nwrite)
+{
+    int ret = ERROR_SUCCESS;
+
+    ssize_t nb_write = st_write(stfd, buf, size, send_timeout);
+    if (nwrite) {
+        *nwrite = nb_write;
+    }
+
+    // On success a non-negative integer equal to nbyte is returned.
+    // Otherwise, a value of -1 is returned and errno is set to indicate the error.
+    if (nb_write <= 0) {
+        // @see https://github.com/simple-rtmp-server/srs/issues/200
+        if (nb_write < 0 && errno == ETIME) {
+            return ERROR_SOCKET_TIMEOUT;
+        }
+
+        return ERROR_SOCKET_WRITE;
+    }
+
+    send_bytes += nb_write;
+
+    return ret;
+}
+
+int DlpStSocket::writev(const iovec *iov, int iov_size, ssize_t* nwrite)
+{
+    int ret = ERROR_SUCCESS;
+
+    ssize_t nb_write = st_writev(stfd, iov, iov_size, send_timeout);
+    if (nwrite) {
+        *nwrite = nb_write;
+    }
+
+    // On success a non-negative integer equal to nbyte is returned.
+    // Otherwise, a value of -1 is returned and errno is set to indicate the error.
+    if (nb_write <= 0) {
+        // @see https://github.com/simple-rtmp-server/srs/issues/200
+        if (nb_write < 0 && errno == ETIME) {
+            return ERROR_SOCKET_TIMEOUT;
+        }
+
+        return ERROR_SOCKET_WRITE;
+    }
+
+    send_bytes += nb_write;
+
+    return ret;
+}
+

trunk/src/core/dlp_core.hpp

@@ -82,6 +82,10 @@ extern int dlp_master_id;
 #define ERROR_ST_OPEN_FD        1003
 #define ERROR_ST_TRHEAD         1004
 #define ERROR_ST_SOCKET         1005
+#define ERROR_SOCKET_TIMEOUT    1006
+#define ERROR_SOCKET_READ       1007
+#define ERROR_SOCKET_READ_FULLY 1008
+#define ERROR_SOCKET_WRITE      1009
 
 // utilies.
 #include <string>
@@ -94,4 +98,40 @@ extern std::string dlp_get_peer_ip(int fd);
 extern void dlp_close_stfd(st_netfd_t& stfd);
 extern int dlp_socket_connect(std::string server, int port, st_utime_t timeout, st_netfd_t* pstfd);
 
+/**
+ * the socket provides TCP socket over st,
+ * that is, the sync socket mechanism.
+ */
+class DlpStSocket
+{
+private:
+    int64_t recv_timeout;
+    int64_t send_timeout;
+    int64_t recv_bytes;
+    int64_t send_bytes;
+    st_netfd_t stfd;
+public:
+    DlpStSocket(st_netfd_t client_stfd);
+    virtual ~DlpStSocket();
+public:
+    virtual bool is_never_timeout(int64_t timeout_us);
+    virtual void set_recv_timeout(int64_t timeout_us);
+    virtual int64_t get_recv_timeout();
+    virtual void set_send_timeout(int64_t timeout_us);
+    virtual int64_t get_send_timeout();
+    virtual int64_t get_recv_bytes();
+    virtual int64_t get_send_bytes();
+public:
+    /**
+     * @param nread, the actual read bytes, ignore if NULL.
+     */
+    virtual int read(void* buf, size_t size, ssize_t* nread);
+    virtual int read_fully(void* buf, size_t size, ssize_t* nread);
+    /**
+     * @param nwrite, the actual write bytes, ignore if NULL.
+     */
+    virtual int write(void* buf, size_t size, ssize_t* nwrite);
+    virtual int writev(const iovec *iov, int iov_size, ssize_t* nwrite);
+};
+
 #endif

trunk/src/core/dlp_core_proxy.cpp

@@ -136,9 +136,52 @@ int DlpProxyConnection::proxy(st_netfd_t srs)
 {
     int ret = ERROR_SUCCESS;
 
-    // TODO: FIXME: implements it.
+    DlpStSocket skt_client(stfd);
+    DlpStSocket skt_srs(srs);
+
+    skt_client.set_recv_timeout(300 * 1000);
+    skt_srs.set_recv_timeout(1500 * 1000);
+
+    char buf[4096];
     for (;;) {
-        st_sleep(3);
+        // proxy client ==> srs.
+        ssize_t nread = 0;
+        for (;;) {
+            nread = 0;
+
+            if ((ret = skt_client.read(buf, 4096, &nread)) != ERROR_SUCCESS) {
+                if (ret != ERROR_SOCKET_TIMEOUT) {
+                    return ret;
+                }
+            }
+
+            if (nread <= 0) {
+                break;
+            }
+
+            if ((ret = skt_srs.write(buf, nread, NULL)) != ERROR_SUCCESS) {
+                return ret;
+            }
+        }
+
+        // proxy srs ==> client
+        for (;;) {
+            nread = 0;
+
+            if ((ret = skt_srs.read(buf, 4096, &nread)) != ERROR_SUCCESS) {
+                if (ret != ERROR_SOCKET_TIMEOUT) {
+                    return ret;
+                }
+            }
+
+            if (nread <= 0) {
+                break;
+            }
+
+            if ((ret = skt_client.write(buf, nread, NULL)) != ERROR_SUCCESS) {
+                return ret;
+            }
+        }
     }
 
     return ret;
@@ -170,6 +213,7 @@ int dlp_connection_proxy(DlpProxyConnection* conn)
     // do proxy.
     ret = conn->proxy(stfd);
     context->release_srs(srs);
+    dlp_close_stfd(stfd);
 
     return ret;
 }