refine code, ignore client when no ip.

@@ -23,6 +23,8 @@ CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
 
 
 #include <srs_app_conn.hpp>
 #include <srs_app_conn.hpp>
 
 
+using namespace std;
+
 #include <srs_kernel_log.hpp>
 #include <srs_kernel_log.hpp>
 #include <srs_kernel_error.hpp>
 #include <srs_kernel_error.hpp>
 #include <srs_app_utility.hpp>
 #include <srs_app_utility.hpp>
@@ -36,11 +38,12 @@ IConnectionManager::~IConnectionManager()
 {
 {
 }
 }
 
 
-SrsConnection::SrsConnection(IConnectionManager* cm, st_netfd_t c)
+SrsConnection::SrsConnection(IConnectionManager* cm, st_netfd_t c, string cip)
 {
 {
     id = 0;
     id = 0;
     manager = cm;
     manager = cm;
     stfd = c;
     stfd = c;
+    ip = cip;
     disposed = false;
     disposed = false;
     expired = false;
     expired = false;
     create_time = srs_get_system_time_ms();
     create_time = srs_get_system_time_ms();
@@ -112,8 +115,6 @@ int SrsConnection::cycle()
     _srs_context->generate_id();
     _srs_context->generate_id();
     id = _srs_context->get_id();
     id = _srs_context->get_id();
     
     
-    ip = srs_get_peer_ip(st_netfd_fileno(stfd));
-    
     int oret = ret = do_cycle();
     int oret = ret = do_cycle();
     
     
     // if socket io error, set to closed.
     // if socket io error, set to closed.

@@ -110,7 +110,7 @@ protected:
      */
      */
     int64_t create_time;
     int64_t create_time;
 public:
 public:
-    SrsConnection(IConnectionManager* cm, st_netfd_t c);
+    SrsConnection(IConnectionManager* cm, st_netfd_t c, std::string cip);
     virtual ~SrsConnection();
     virtual ~SrsConnection();
 // interface IKbpsDelta
 // interface IKbpsDelta
 public:
 public:

@@ -1309,8 +1309,8 @@ int SrsGoApiError::serve_http(ISrsHttpResponseWriter* w, ISrsHttpMessage* r)
     return srs_api_response_code(w, r, 100);
     return srs_api_response_code(w, r, 100);
 }
 }
 
 
-SrsHttpApi::SrsHttpApi(IConnectionManager* cm, st_netfd_t fd, SrsHttpServeMux* m)
-    : SrsConnection(cm, fd)
+SrsHttpApi::SrsHttpApi(IConnectionManager* cm, st_netfd_t fd, SrsHttpServeMux* m, string cip)
+    : SrsConnection(cm, fd, cip)
 {
 {
     mux = m;
     mux = m;
     parser = new SrsHttpParser();
     parser = new SrsHttpParser();

@@ -215,7 +215,7 @@ private:
     bool crossdomain_required;
     bool crossdomain_required;
     bool crossdomain_enabled;
     bool crossdomain_enabled;
 public:
 public:
-    SrsHttpApi(IConnectionManager* cm, st_netfd_t fd, SrsHttpServeMux* m);
+    SrsHttpApi(IConnectionManager* cm, st_netfd_t fd, SrsHttpServeMux* m, std::string cip);
     virtual ~SrsHttpApi();
     virtual ~SrsHttpApi();
 // interface IKbpsDelta
 // interface IKbpsDelta
 public:
 public:

@@ -1063,8 +1063,8 @@ int SrsHttpParser::on_body(http_parser* parser, const char* at, size_t length)
     return 0;
     return 0;
 }
 }
 
 
-SrsHttpConn::SrsHttpConn(IConnectionManager* cm, st_netfd_t fd, ISrsHttpServeMux* m)
-    : SrsConnection(cm, fd)
+SrsHttpConn::SrsHttpConn(IConnectionManager* cm, st_netfd_t fd, ISrsHttpServeMux* m, string cip)
+    : SrsConnection(cm, fd, cip)
 {
 {
     parser = new SrsHttpParser();
     parser = new SrsHttpParser();
     http_mux = m;
     http_mux = m;
@@ -1187,8 +1187,8 @@ int SrsHttpConn::on_disconnect(SrsRequest* req)
     return ret;
     return ret;
 }
 }
 
 
-SrsResponseOnlyHttpConn::SrsResponseOnlyHttpConn(IConnectionManager* cm, st_netfd_t fd, ISrsHttpServeMux* m)
-    : SrsHttpConn(cm, fd, m)
+SrsResponseOnlyHttpConn::SrsResponseOnlyHttpConn(IConnectionManager* cm, st_netfd_t fd, ISrsHttpServeMux* m, string cip)
+    : SrsHttpConn(cm, fd, m, cip)
 {
 {
 }
 }
 
 

@@ -350,7 +350,7 @@ private:
     SrsHttpParser* parser;
     SrsHttpParser* parser;
     ISrsHttpServeMux* http_mux;
     ISrsHttpServeMux* http_mux;
 public:
 public:
-    SrsHttpConn(IConnectionManager* cm, st_netfd_t fd, ISrsHttpServeMux* m);
+    SrsHttpConn(IConnectionManager* cm, st_netfd_t fd, ISrsHttpServeMux* m, std::string cip);
     virtual ~SrsHttpConn();
     virtual ~SrsHttpConn();
 // interface IKbpsDelta
 // interface IKbpsDelta
 public:
 public:
@@ -381,7 +381,7 @@ private:
 class SrsResponseOnlyHttpConn : public SrsHttpConn
 class SrsResponseOnlyHttpConn : public SrsHttpConn
 {
 {
 public:
 public:
-    SrsResponseOnlyHttpConn(IConnectionManager* cm, st_netfd_t fd, ISrsHttpServeMux* m);
+    SrsResponseOnlyHttpConn(IConnectionManager* cm, st_netfd_t fd, ISrsHttpServeMux* m, std::string cip);
     virtual ~SrsResponseOnlyHttpConn();
     virtual ~SrsResponseOnlyHttpConn();
 public:
 public:
     virtual int on_got_http_message(ISrsHttpMessage* msg);
     virtual int on_got_http_message(ISrsHttpMessage* msg);

@@ -312,11 +312,11 @@ void SrsSimpleRtmpClient::set_recv_timeout(int64_t timeout)
 }
 }
 
 
 #ifdef SRS_AUTO_KAFKA
 #ifdef SRS_AUTO_KAFKA
-SrsRtmpConn::SrsRtmpConn(SrsServer* svr, ISrsKafkaCluster* k, st_netfd_t c)
+SrsRtmpConn::SrsRtmpConn(SrsServer* svr, ISrsKafkaCluster* k, st_netfd_t c, string cip)
 #else
 #else
-SrsRtmpConn::SrsRtmpConn(SrsServer* svr, st_netfd_t c)
+SrsRtmpConn::SrsRtmpConn(SrsServer* svr, st_netfd_t c, string cip)
 #endif
 #endif
-    : SrsConnection(svr, c)
+    : SrsConnection(svr, c, cip)
 {
 {
     server = svr;
     server = svr;
 #ifdef SRS_AUTO_KAFKA
 #ifdef SRS_AUTO_KAFKA
@@ -373,7 +373,7 @@ int SrsRtmpConn::do_cycle()
 {
 {
     int ret = ERROR_SUCCESS;
     int ret = ERROR_SUCCESS;
     
     
-    srs_trace("RTMP client ip=%s", ip.c_str());
+    srs_trace("RTMP client ip=%s, fd=%d", ip.c_str(), st_netfd_fileno(stfd));
     
     
     // notify kafka cluster.
     // notify kafka cluster.
 #ifdef SRS_AUTO_KAFKA
 #ifdef SRS_AUTO_KAFKA

@@ -144,9 +144,9 @@ private:
 #endif
 #endif
 public:
 public:
 #ifdef SRS_AUTO_KAFKA
 #ifdef SRS_AUTO_KAFKA
-    SrsRtmpConn(SrsServer* svr, ISrsKafkaCluster* k, st_netfd_t c);
+    SrsRtmpConn(SrsServer* svr, ISrsKafkaCluster* k, st_netfd_t c, std::string cip);
 #else
 #else
-    SrsRtmpConn(SrsServer* svr, st_netfd_t c);
+    SrsRtmpConn(SrsServer* svr, st_netfd_t c, std::string cip);
 #endif
 #endif
     virtual ~SrsRtmpConn();
     virtual ~SrsRtmpConn();
 public:
 public:

@@ -1241,25 +1241,55 @@ void SrsServer::resample_kbps()
     srs_update_rtmp_server((int)conns.size(), kbps);
     srs_update_rtmp_server((int)conns.size(), kbps);
 }
 }
 
 
-int SrsServer::accept_client(SrsListenerType type, st_netfd_t client_stfd)
+int SrsServer::accept_client(SrsListenerType type, st_netfd_t stfd)
 {
 {
     int ret = ERROR_SUCCESS;
     int ret = ERROR_SUCCESS;
     
     
-    int fd = st_netfd_fileno(client_stfd);
+    SrsConnection* conn = fd2conn(type, stfd);
+    if (conn == NULL) {
+        srs_close_stfd(stfd);
+        return ERROR_SUCCESS;
+    }
+    srs_assert(conn);
+    
+    // directly enqueue, the cycle thread will remove the client.
+    conns.push_back(conn);
+    srs_verbose("add conn to vector.");
+    
+    // cycle will start process thread and when finished remove the client.
+    // @remark never use the conn, for it maybe destroyed.
+    if ((ret = conn->start()) != ERROR_SUCCESS) {
+        return ret;
+    }
+    srs_verbose("accept client finished. conns=%d, ret=%d", (int)conns.size(), ret);
+    
+    return ret;
+}
+
+SrsConnection* SrsServer::fd2conn(SrsListenerType type, st_netfd_t stfd)
+{
+    int ret = ERROR_SUCCESS;
+    
+    int fd = st_netfd_fileno(stfd);
+    string ip = srs_get_peer_ip(fd);
+    
+    // for some keep alive application, for example, the keepalived,
+    // will send some tcp packet which we cann't got the ip,
+    // we just ignore it.
+    if (ip.empty()) {
+        srs_info("ignore empty ip client, fd=%d.", fd);
+        return NULL;
+    }
 
 
     // check connection limitation.
     // check connection limitation.
     int max_connections = _srs_config->get_max_connections();
     int max_connections = _srs_config->get_max_connections();
     if (handler && (ret = handler->on_accept_client(max_connections, (int)conns.size()) != ERROR_SUCCESS)) {
     if (handler && (ret = handler->on_accept_client(max_connections, (int)conns.size()) != ERROR_SUCCESS)) {
-        srs_error("handle accept client failed, drop client: "
-            "clients=%d, max=%d, fd=%d. ret=%d", (int)conns.size(), max_connections, fd, ret);
-        srs_close_stfd(client_stfd);
-        return ERROR_SUCCESS;
+        srs_error("handle accept client failed, drop client: clients=%d, max=%d, fd=%d. ret=%d", (int)conns.size(), max_connections, fd, ret);
+        return NULL;
     }
     }
     if ((int)conns.size() >= max_connections) {
     if ((int)conns.size() >= max_connections) {
-        srs_error("exceed the max connections, drop client: "
-            "clients=%d, max=%d, fd=%d", (int)conns.size(), max_connections, fd);
-        srs_close_stfd(client_stfd);
-        return ERROR_SUCCESS;
+        srs_error("exceed the max connections, drop client: clients=%d, max=%d, fd=%d", (int)conns.size(), max_connections, fd);
+        return NULL;
     }
     }
     
     
     // avoid fd leak when fork.
     // avoid fd leak when fork.
@@ -1269,56 +1299,40 @@ int SrsServer::accept_client(SrsListenerType type, st_netfd_t client_stfd)
         if ((val = fcntl(fd, F_GETFD, 0)) < 0) {
         if ((val = fcntl(fd, F_GETFD, 0)) < 0) {
             ret = ERROR_SYSTEM_PID_GET_FILE_INFO;
             ret = ERROR_SYSTEM_PID_GET_FILE_INFO;
             srs_error("fnctl F_GETFD error! fd=%d. ret=%#x", fd, ret);
             srs_error("fnctl F_GETFD error! fd=%d. ret=%#x", fd, ret);
-            srs_close_stfd(client_stfd);
-            return ret;
+            return NULL;
         }
         }
         val |= FD_CLOEXEC;
         val |= FD_CLOEXEC;
         if (fcntl(fd, F_SETFD, val) < 0) {
         if (fcntl(fd, F_SETFD, val) < 0) {
             ret = ERROR_SYSTEM_PID_SET_FILE_INFO;
             ret = ERROR_SYSTEM_PID_SET_FILE_INFO;
             srs_error("fcntl F_SETFD error! fd=%d ret=%#x", fd, ret);
             srs_error("fcntl F_SETFD error! fd=%d ret=%#x", fd, ret);
-            srs_close_stfd(client_stfd);
-            return ret;
+            return NULL;
         }
         }
     }
     }
     
     
     SrsConnection* conn = NULL;
     SrsConnection* conn = NULL;
+    
     if (type == SrsListenerRtmpStream) {
     if (type == SrsListenerRtmpStream) {
-        conn = new SrsRtmpConn(this, kafka, client_stfd);
+        conn = new SrsRtmpConn(this, kafka, stfd, ip);
     } else if (type == SrsListenerHttpApi) {
     } else if (type == SrsListenerHttpApi) {
 #ifdef SRS_AUTO_HTTP_API
 #ifdef SRS_AUTO_HTTP_API
-        conn = new SrsHttpApi(this, client_stfd, http_api_mux);
+        conn = new SrsHttpApi(this, stfd, http_api_mux, ip);
 #else
 #else
         srs_warn("close http client for server not support http-api");
         srs_warn("close http client for server not support http-api");
-        srs_close_stfd(client_stfd);
+        srs_close_stfd(stfd);
         return ret;
         return ret;
 #endif
 #endif
     } else if (type == SrsListenerHttpStream) {
     } else if (type == SrsListenerHttpStream) {
 #ifdef SRS_AUTO_HTTP_SERVER
 #ifdef SRS_AUTO_HTTP_SERVER
-        conn = new SrsResponseOnlyHttpConn(this, client_stfd, http_server);
+        conn = new SrsResponseOnlyHttpConn(this, stfd, http_server, ip);
 #else
 #else
         srs_warn("close http client for server not support http-server");
         srs_warn("close http client for server not support http-server");
-        srs_close_stfd(client_stfd);
-        return ret;
+        return NULL;
 #endif
 #endif
     } else {
     } else {
         // TODO: FIXME: handler others
         // TODO: FIXME: handler others
     }
     }
-    srs_assert(conn);
     
     
-    // directly enqueue, the cycle thread will remove the client.
-    conns.push_back(conn);
-    srs_verbose("add conn to vector.");
-    
-    // cycle will start process thread and when finished remove the client.
-    // @remark never use the conn, for it maybe destroyed.
-    if ((ret = conn->start()) != ERROR_SUCCESS) {
-        return ret;
-    }
-    srs_verbose("conn started success.");
-
-    srs_verbose("accept client finished. conns=%d, ret=%d", (int)conns.size(), ret);
-    
-    return ret;
+    return conn;
 }
 }
 
 
 void SrsServer::remove(SrsConnection* conn)
 void SrsServer::remove(SrsConnection* conn)

@@ -365,9 +365,11 @@ public:
     * when listener got a fd, notice server to accept it.
     * when listener got a fd, notice server to accept it.
     * @param type, the client type, used to create concrete connection, 
     * @param type, the client type, used to create concrete connection, 
     *       for instance RTMP connection to serve client.
     *       for instance RTMP connection to serve client.
-    * @param client_stfd, the client fd in st boxed, the underlayer fd.
+    * @param stfd, the client fd in st boxed, the underlayer fd.
     */
     */
-    virtual int accept_client(SrsListenerType type, st_netfd_t client_stfd);
+    virtual int accept_client(SrsListenerType type, st_netfd_t stfd);
+private:
+    virtual SrsConnection* fd2conn(SrsListenerType type, st_netfd_t stfd);
 // IConnectionManager
 // IConnectionManager
 public:
 public:
     /**
     /**