GithubStar/neuron
1
0
Fork 0
mirror of https://github.com/emqx/neuron.git synced 2026-10-29 07:06:32 +08:00
Code Issues Packages Projects Releases Wiki Activity

improve(connection): replace old client

Browse Source
This commit is contained in:
fengzero
2023-04-10 03:12:27 +00:00
parent 6c24b78781
commit 7548de61c6
1 changed files with 37 additions and 11 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/connection/connection.c
+37 -11
View File
@@ -69,6 +69,7 @@ struct neu_conn {
static void conn_tcp_server_add_client(neu_conn_t *conn, int fd,
struct sockaddr_in client);
static void conn_tcp_server_del_client(neu_conn_t *conn, int fd);
static void conn_tcp_server_replace_client(neu_conn_t *conn, int fd);
static void conn_tcp_server_listen(neu_conn_t *conn);
static void conn_tcp_server_stop(neu_conn_t *conn);
@@ -180,16 +181,6 @@ int neu_conn_tcp_server_accept(neu_conn_t *conn)
return -1;
}
if (conn->tcp_server.n_client >= conn->param.params.tcp_server.max_link) {
close(fd);
zlog_warn(conn->param.log, "%s:%d accpet max link: %d, reject",
conn->param.params.tcp_server.ip,
conn->param.params.tcp_server.port,
conn->param.params.tcp_server.max_link);
pthread_mutex_unlock(&conn->mtx);
return -1;
}
if (conn->block) {
struct timeval tv = {
.tv_sec = conn->param.params.tcp_server.timeout / 1000,
@@ -199,7 +190,23 @@ int neu_conn_tcp_server_accept(neu_conn_t *conn)
setsockopt(fd, SOL_SOCKET, SO_SNDTIMEO, &tv, sizeof(tv));
}
conn_tcp_server_add_client(conn, fd, client);
if (conn->tcp_server.n_client >= conn->param.params.tcp_server.max_link) {
int free_fd = conn_tcp_server_del_first_client(conn, fd);
if (free_fd > 0) {
zlog_warn(conn->param.log, "replace old client %d with %d", free_fd,
fd);
} else {
close(fd);
zlog_warn(conn->param.log, "%s:%d accpet max link: %d, reject",
conn->param.params.tcp_server.ip,
conn->param.params.tcp_server.port,
conn->param.params.tcp_server.max_link);
pthread_mutex_unlock(&conn->mtx);
return -1;
}
} else {
conn_tcp_server_add_client(conn, fd, client);
}
conn->is_connected = true;
conn->connected(conn->data, fd);
@@ -891,6 +898,25 @@ static void conn_tcp_server_del_client(neu_conn_t *conn, int fd)
}
}
static void conn_tcp_server_replace_client(neu_conn_t *conn, int fd)
{
int ret = 0;
for (int i = 0; i < conn->param.params.tcp_server.max_link; i++) {
if (conn->tcp_server.clients[i].fd > 0) {
ret = conn->tcp_server.clients[i].fd;
close(conn->tcp_server.clients[i].fd);
conn->tcp_server.clients[i].fd = 0;
conn->tcp_server.clients[i].fd = fd;
return;
}
}
return ret;
}
int neu_conn_stream_consume(neu_conn_t *conn, void *context,
neu_conn_stream_consume_fn fn)
{