libwebsockets 如何添加其他socket事件回调

Question

功能逻辑需求如下：
客户端A通过tcp socket长连接给中间服务器B发送消息， 中间服务器B通过libwebsockets库采用ws协议向 WS服务器C 转发数据：

中间服务器B代码如下： 在单线程中，先监听一个tcp socket，等待客户端A连上来， 客户端A会不定时往 tcp_fd 发送数据，B要怎样将tcp_fd的收发事件加入到libwebsockets的事件列表中，
使得B在while ( 1 ) {}循环里面既能处理ws的callback，又能及时处理 tcp_fd 的收发数据？ 如果每次等到 lws_service(context, 5) 超时出来之后，再处理tcp_fd的数据，感觉不够及时，对libwebsockets不熟悉，
不知道怎样添加自定义socket读写事件到libwebsockets中，请教一下大家

int callback(struct lws *wsi, enum lws_callback_reasons reason, void *user, void *in, size_t len) {
    struct session_data *data = (struct session_data *) user;
    switch (reason) {
    case LWS_CALLBACK_CLIENT_ESTABLISHED:  
        lwsl_notice("Connect ok!\n");
        break;

    case LWS_CALLBACK_CLIENT_RECEIVE:   
        break;
    case LWS_CALLBACK_CLIENT_WRITEABLE:   
        break;
    }
    return 0;
}


struct lws_protocols protocols[] = {
    {
        "ws", callback, sizeof(struct session_data), MAX_PAYLOAD_SIZE,
    },
    {
        NULL, NULL,   0 
    }
};


int main() {
    int listenfd, tcp_fd;
    struct sockaddr_in server_addr, client_addr;
    socklen_t sin_size;
    char buffer[1024];
    int n;

    listenfd = socket(AF_INET, SOCK_STREAM, 0);

    memset(&server_addr, 0, sizeof(server_addr));
    server_addr.sin_family = AF_INET;
    server_addr.sin_addr.s_addr = INADDR_ANY; // 监听所有IP地址的连接请求
    server_addr.sin_port = htons(8080); // 端口号8080
    if (bind(listenfd, (struct sockaddr *)&server_addr, sizeof(server_addr)) == -1) {
        perror("bind");
        exit(1);
    }
    listen(listenfd, 5); // 监听5个连接请求
    sin_size = sizeof(client_addr); // 客户端地址结构的大小
    tcp_fd = accept(listenfd, (struct sockaddr *)&client_addr, &sin_size); // 接受连接请求并获取新的文件描述符
    
........................
........................


    struct lws_context_creation_info ctx_info = { 0 };
    ctx_info.port = CONTEXT_PORT_NO_LISTEN;
    ctx_info.iface = NULL;
    ctx_info.protocols = protocols;
    ctx_info.gid = -1;
    ctx_info.uid = -1;

    struct lws_context *context = lws_create_context(&ctx_info);

    char address[] = "abc.com";
    int port = 443;
    char addr_str[256] = { 0 };
    sprintf(addr_str, "%s", address );

    struct lws_client_connect_info conn_info = { 0 };
    conn_info.context = context;
    conn_info.address = address;
    conn_info.port = port;
    conn_info.ssl_connection = 1;
    conn_info.path = "/";
    conn_info.host = addr_str;
    conn_info.origin = addr_str;
    conn_info.protocol = protocols[0].name;

    struct lws *wsi = lws_client_connect_via_info(&conn_info);

    while ( 1 ) {
        lws_service(context, 5);
        lws_callback_on_writable(wsi);
    }

    lws_context_destroy(context);

    return 0;
}

Answer 1

此答案是由GPT4和本人亲自作答，如有帮助，还请采纳！
要解决你提出的需求，我们需要在 libwebsockets 的事件循环中同时处理 WebSocket 连接（lws_service）和 TCP socket 连接（tcp_fd）的事件。libwebsockets 是一个事件驱动的库，它的事件循环 lws_service 会阻塞并处理各种事件，包括 WebSocket 的回调。你的需求是将 TCP socket 的事件（如读取和写入）集成到这个事件循环中，以实现及时的数据收发，而不是依赖于 lws_service 的超时轮询。

思路和解决方案：

1. 使用 libwebsockets 的事件循环：libwebsockets 自身并不直接支持将外部的 socket 事件（比如 TCP socket）集成到它的事件循环中，但它支持扩展事件循环，通过 libevent、libev 或 poll 等方式实现多路复用。你可以在主线程中使用这些方式来同时监听 tcp_fd 和 WebSocket 事件。

2. 使用 libwebsockets 的 lws_fd_to_context 功能：libwebsockets 提供了 lws_fd_to_context 和 lws_callback_on_writable 等 API，允许你手动通知库某个 socket 可读或可写。

3. 使用 select 或 poll 来管理多个 socket 事件：通过 select 或 poll 机制，结合 lws_service 来处理多路复用，使得 TCP 连接的数据能够及时处理。

具体实现方案：

1. 整合 libwebsockets 的事件循环与 select：

你可以利用 select 或 poll 来并行监听多个 socket，包括你的 TCP socket 和 WebSocket 连接。lws_service 会调用一次 select，等待 WebSocket 数据，并且能够在其中插入对 TCP socket 的监听。

2. 代码实现：

#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <unistd.h>
#include <sys/socket.h>
#include <arpa/inet.h>
#include <poll.h>
#include <libwebsockets.h>

// WebSocket回调函数
int callback(struct lws *wsi, enum lws_callback_reasons reason, void *user, void *in, size_t len) {
    struct session_data *data = (struct session_data *) user;
    switch (reason) {
        case LWS_CALLBACK_CLIENT_ESTABLISHED:
            lwsl_notice("WebSocket connection established.\n");
            break;
        case LWS_CALLBACK_CLIENT_RECEIVE:
            lwsl_notice("Received WebSocket data: %s\n", (char *) in);
            break;
        case LWS_CALLBACK_CLIENT_WRITEABLE:
            lwsl_notice("WebSocket is writable.\n");
            break;
        default:
            break;
    }
    return 0;
}

// WebSocket协议
struct lws_protocols protocols[] = {
    {
        "ws", callback, sizeof(struct session_data), 1024,
    },
    { NULL, NULL, 0 } // 结束符
};

int main() {
    int listenfd, tcp_fd;
    struct sockaddr_in server_addr, client_addr;
    socklen_t sin_size;
    char buffer[1024];
    int n;

    // 创建监听 socket
    listenfd = socket(AF_INET, SOCK_STREAM, 0);
    memset(&server_addr, 0, sizeof(server_addr));
    server_addr.sin_family = AF_INET;
    server_addr.sin_addr.s_addr = INADDR_ANY;
    server_addr.sin_port = htons(8080); // 监听端口 8080

    if (bind(listenfd, (struct sockaddr *)&server_addr, sizeof(server_addr)) == -1) {
        perror("bind");
        exit(1);
    }

    listen(listenfd, 5);
    sin_size = sizeof(client_addr);
    tcp_fd = accept(listenfd, (struct sockaddr *)&client_addr, &sin_size);

    // WebSocket初始化
    struct lws_context_creation_info ctx_info = { 0 };
    ctx_info.port = CONTEXT_PORT_NO_LISTEN;
    ctx_info.protocols = protocols;
    ctx_info.gid = -1;
    ctx_info.uid = -1;

    struct lws_context *context = lws_create_context(&ctx_info);
    
    // WebSocket连接信息
    struct lws_client_connect_info conn_info = { 0 };
    conn_info.context = context;
    conn_info.address = "abc.com";
    conn_info.port = 443;
    conn_info.ssl_connection = 1;
    conn_info.path = "/";
    conn_info.host = "abc.com";
    conn_info.origin = "abc.com";
    conn_info.protocol = protocols[0].name;
    
    struct lws *wsi = lws_client_connect_via_info(&conn_info);

    // 设置pollfd数组
    struct pollfd fds[2];
    fds[0].fd = tcp_fd;
    fds[0].events = POLLIN;
    fds[1].fd = lws_get_socket_fd(wsi);  // WebSocket的socket文件描述符
    fds[1].events = POLLIN | POLLOUT; // 可读可写

    while (1) {
        int nready = poll(fds, 2, 5);  // 设置5ms超时

        if (nready < 0) {
            perror("poll");
            break;
        }

        if (fds[0].revents & POLLIN) {
            // 处理TCP socket接收的数据
            n = read(tcp_fd, buffer, sizeof(buffer));
            if (n > 0) {
                lwsl_notice("Received from TCP: %s\n", buffer);
                // 将数据转发到WebSocket
                lws_write(wsi, (unsigned char *) buffer, n, LWS_WRITE_TEXT);
            }
        }

        if (fds[1].revents & POLLIN) {
            // 处理WebSocket接收的数据
            lws_service(context, 0);
        }

        if (fds[1].revents & POLLOUT) {
            // 处理WebSocket可写事件
            lws_callback_on_writable(wsi);
        }
    }

    // 清理资源
    lws_context_destroy(context);
    close(tcp_fd);
    close(listenfd);
    return 0;
}

代码解析：

1. TCP Socket 监听：

   • 使用 socket 创建监听套接字 listenfd，并绑定到指定的 IP 地址和端口。
   • 使用 accept 接受客户端的连接请求，返回新的 TCP 连接套接字 tcp_fd。

2. WebSocket 连接：

   • 使用 libwebsockets 初始化 WebSocket 客户端上下文，并通过 lws_client_connect_via_info 连接到 WebSocket 服务器。
   • 使用 lws_get_socket_fd(wsi) 获取 WebSocket 的套接字文件描述符。

3. 使用 poll 实现多路复用：

   • 使用 poll 来监听 tcp_fd 和 WebSocket 套接字的事件。
   • 如果 tcp_fd 可读，则读取数据并将其转发到 WebSocket。
   • 如果 WebSocket 可读，则调用 lws_service 处理 WebSocket 数据。
   • 如果 WebSocket 可写，则调用 lws_callback_on_writable 进行处理。

4. 事件回调：

   • WebSocket 的回调函数 callback 处理 WebSocket 连接建立、数据接收和可写事件。

关键点总结：

• poll 用于多路复用：通过 poll 同时监听 TCP socket 和 WebSocket 事件，确保能及时处理 TCP 数据。
• **lws_service 结合 poll**：在事件循环中，lws_service 负责处理 WebSocket 的事件，并确保 WebSocket 能够正常读写数据。
• WebSocket 和 TCP 数据交互：通过读取 TCP 数据并转发到 WebSocket，确保 WebSocket 服务器能够收到来自 TCP 客户端的数据。

进一步优化：

• 增加错误处理：可以增加更多的错误检查，尤其是在 socket 连接和数据处理过程中。
• 高性能优化：如果需要高并发处理，可以考虑使用 epoll 或 kqueue 替代 poll，以提高性能。

通过这种方式，你可以在 libwebsockets 的事件循环中同时处理 WebSocket 和 TCP socket 的数据，避免了延迟处理问题。