[fix]udp runtime error

include/udp_client.h

@@ -3,7 +3,6 @@
 
 #include "hv/hloop.h"
 #include "hv/UdpClient.h"
-#include "timer_thread.h"
 #include "bolt/crypt.h"
 #include "udp_client_map.h"
 #include <string>
@@ -26,23 +25,24 @@ class UdpClientProxy
 {
 public:
     UdpClientProxy(struct sockaddr_in t_addr, struct sockaddr_in u_addr, uint32_t session_id) : t_addr(t_addr), u_addr(u_addr), session_id(session_id) {}
-    hv::UdpClient *getClient() { return &cli; }
-    bool init();
+    hio_t *getClientIO() { return cli_io; }
+    bool init(hloop_t* loop);
     void close();
     bool isDns() {
         return ntohs(t_addr.sin_port) == 53;
     };
 
 private:
-    void onRecv(const hv::SocketChannelPtr &channel, hv::Buffer *buf);
-    void onWrited(const hv::SocketChannelPtr &channel, hv::Buffer *buf);
+    void onRecv(void* buf, int readbytes);
+    void onWrited(const void* buf, int writebytes);
+    void onClosed();
 
 public:
     struct sockaddr_in t_addr, u_addr;
 
 private:
-    hv::UdpClient cli;
-    TimerThread timer;
+    hio_t *cli_io;
+    htimer_t* timer;
     uint32_t session_id;
     long long ts;
 };

main.cpp

@@ -75,6 +75,7 @@ static HTHREAD_RETTYPE udpbind_thread(void *userdata)
     }
     int bindfd = hio_fd(bindio);
     spdlog::info("ProxyServer start: udp->{}", port);
+    hevent_set_userdata(bindio, loop); //必须
     hio_setcb_read(bindio, udp_on_recvfrom);
     hio_setcb_write(bindio, udp_on_writed);
     hio_setcb_close(bindio, udp_on_close);
@@ -113,7 +114,6 @@ int main(int argc, char **argv)
     if (type == 1)
     {
         // Udp
-        spdlog::info("ProxyServer start: udp->{}", port);
         udpbind_loop = hloop_new(HLOOP_FLAG_AUTO_FREE);
         udpbind_thread(udpbind_loop);
     }

src/tcp_client.cpp

@@ -92,7 +92,7 @@ bool TcpClientBolt::handShake(void *buf, int readbytes)
                     
                     if (!this->connect((struct sockaddr *)&t_addr))
                     {
-                        spdlog::info("BOLT_CHANNEL_CMD_TCP_HANDSHAKE_REQUEST: t_addr={}:{}, u_addr={}:{} ==> connect fail", inet_ntoa(t_addr.sin_addr), htons(t_addr.sin_port), inet_ntoa(u_addr.sin_addr), htons(u_addr.sin_port));
+                        spdlog::info("[TCP]BOLT_CHANNEL_CMD_TCP_HANDSHAKE_REQUEST: t_addr={}:{}, u_addr={}:{} ==> connect fail", inet_ntoa(t_addr.sin_addr), htons(t_addr.sin_port), inet_ntoa(u_addr.sin_addr), htons(u_addr.sin_port));
                         char buff[2];
                         buff[0] = BOLT_CHANNEL_CMD_TCP_HANDSHAKE_RESPONSE;
                         buff[1] = BOLT_TCP_HANDSHAKE_FAIL_TIMEOUT;
@@ -124,7 +124,7 @@ bool TcpClientBolt::handShake(void *buf, int readbytes)
                     config.ept_type = config.encrypt ? CRYPT_TYPE::XOR : CRYPT_TYPE::NONE;
                     config.ept_key = config.encrypt ? generateRandomKey() : 0;
 
-                    spdlog::info("BOLT_CHANNEL_CMD_BIND_REQUEST: requestId:{}, session_id:{}, encrypt:{}", request_id, session_id, config.encrypt);
+                    spdlog::info("[TCP]BOLT_CHANNEL_CMD_BIND_REQUEST: requestId:{}, session_id:{}, encrypt:{}", request_id, session_id, config.encrypt);
 
                     GENERATE_DECRYPT_KEY(extend_response, extend_response_len, config.ept_type, config.ept_key)
 
@@ -145,7 +145,7 @@ bool TcpClientBolt::handShake(void *buf, int readbytes)
                     TcpClientMap<uint32_t, tcpBoltConfig>::getInstance().remove(session_id);
                     this->close();
                     hio_close_async(io);
-                    spdlog::info("BOLT_CHANNEL_CMD_UNBIND_REQUEST: t_addr={}:{}, u_addr={}:{}", inet_ntoa(t_addr.sin_addr), htons(t_addr.sin_port), inet_ntoa(u_addr.sin_addr), htons(u_addr.sin_port));
+                    spdlog::info("[TCP]BOLT_CHANNEL_CMD_UNBIND_REQUEST: requestId:{}, session_id:{} t_addr={}:{}, u_addr={}:{}", config->request_id, config->session_id, inet_ntoa(t_addr.sin_addr), htons(t_addr.sin_port), inet_ntoa(u_addr.sin_addr), htons(u_addr.sin_port));
                 }
             }
         }
@@ -179,14 +179,14 @@ bool TcpClientBolt::connect(struct sockaddr *addr)
 
     cli.onMessage = [this](const hv::SocketChannelPtr &channel, hv::Buffer *buf)
     {
-        spdlog::info("<<< len:{} : {}}\n", (int)buf->size(), (char *)buf->data());
+        spdlog::info("<<< len:{} : {}", (int)buf->size(), (char *)buf->data());
 
         onRecv(channel, buf);
     };
 
     cli.onWriteComplete = [this](const hv::SocketChannelPtr &channel, hv::Buffer *buf)
     {
-        spdlog::info(">>> len:{} : {}}\n", (int)buf->size(), (char *)buf->data());
+        spdlog::info(">>> len:{} : {}", (int)buf->size(), (char *)buf->data());
 
         onWrited(channel, buf);
     };
@@ -225,5 +225,6 @@ int TcpClientBolt::send(char *data, int size)
 
 void TcpClientBolt::close()
 {
-    cli.closesocket();
+    spdlog::info("TcpClientBolt::close()\n");
+    cli.stop();
 }

src/tcp_inbound.cpp

@@ -11,8 +11,8 @@ static void tcp_on_close(hio_t* io) {
 
     auto cli = TcpConnMap<hio_t*, TcpClientBolt>::getInstance().get(io);
     if(cli) {
-        TcpConnMap<hio_t*, TcpClientBolt>::getInstance().remove(io);
         cli->close();
+        TcpConnMap<hio_t*, TcpClientBolt>::getInstance().remove(io);
     }
 }
 

src/udp_client.cpp

@@ -21,13 +21,14 @@ bool UdpServerBoltProxy::analyzeData(struct sockaddr_in t_addr, struct sockaddr_
     auto client = _map.get(key);
     if (!client)
     {
+        hloop_t *loop = (hloop_t *)hevent_userdata(io);
         std::unique_ptr<UdpClientProxy> new_client = std::make_unique<UdpClientProxy>(t_addr, u_addr, session_id);
-        if (!new_client->init())
+        if (!new_client->init(loop))
         {
             return false;
         }
-        _map.add(key, new_client);
         client = new_client.get();
+        _map.add(key, new_client);
     }
 
     // if (getConfig().ept_type == CRYPT_TYPE::XOR)
@@ -35,7 +36,7 @@ bool UdpServerBoltProxy::analyzeData(struct sockaddr_in t_addr, struct sockaddr_
     //     xor_::crypt((char *)dest, dest_len, getConfig().ept_key);
     // }
 
-    client->getClient()->sendto(dest, dest_len);
+    hio_write(client->getClientIO(), dest, dest_len);
 
     return true;
 }
@@ -55,13 +56,7 @@ int UdpServerBoltProxy::sendData(struct sockaddr_in t_addr, struct sockaddr_in u
 void UdpServerBoltProxy::closeClient(struct sockaddr_in &t_addr, struct sockaddr_in &u_addr)
 {
     auto key = get_key(t_addr, u_addr);
-    auto client = _map.get(key);
-    if (client)
-    {
-        client->close();
-        _map.remove(key);
-    }
-    
+    _map.remove(key);
 }
 
 void UdpServerBoltProxy::recycle()
@@ -73,70 +68,106 @@ void UdpServerBoltProxy::recycle()
     _map.clear();
 }
 
-bool UdpClientProxy::init()
+bool UdpClientProxy::init(hloop_t *loop)
 {
 
-    int sockfd = cli.createsocket(t_addr.sin_port, inet_ntoa(t_addr.sin_addr));
-    if (sockfd < 0)
+    spdlog::info("UdpClientProxy::init {}:{}<==>{}:{}", inet_ntoa(u_addr.sin_addr), ntohs(u_addr.sin_port), inet_ntoa(t_addr.sin_addr), ntohs(t_addr.sin_port));
+    cli_io = hloop_create_udp_client(loop, inet_ntoa(t_addr.sin_addr), ntohs(t_addr.sin_port));
+    if (cli_io == nullptr)
     {
         return false;
     }
-    cli.onWriteComplete = [this](const hv::SocketChannelPtr &channel, hv::Buffer *buf)
-    {
-        onWrited(channel, buf);
-    };
-    cli.onMessage = [this](const hv::SocketChannelPtr &channel, hv::Buffer *buf)
-    {
-        onRecv(channel, buf);
-    };
-    hio_setcb_close(cli.channel.get()->io(), [](hio_t *io) {
-
+    hevent_set_userdata(cli_io, this);
+    hio_setcb_read(cli_io, [](hio_t *io, void *buf, int readbytes) {
+        UdpClientProxy *cli = (UdpClientProxy *)hevent_userdata(io);
+        if (cli)
+       {
+           cli->onRecv(buf, readbytes);
+       } 
     });
-    cli.start();
-
-    timer.setInterval(TIMEOUT_TIMEUNIT, [this](hv::TimerID timerID) {
-        long timeout = isDns() ? DNS_TIMEOUT : TIMEOUT_TIMEUNIT;
-        long time = currentTimeMillis() - ts;
-        if (time > 0 && time < timeout) 
+    hio_setcb_write(cli_io, [](hio_t *io, const void *buf, int writebytes) {
+        UdpClientProxy* cli = (UdpClientProxy*)hevent_userdata(io);
+        if (cli)
+        {
+            cli->onWrited(buf, writebytes);
+        } 
+    });
+    hio_setcb_close(cli_io, [](hio_t *io) {
+        UdpClientProxy* cli = (UdpClientProxy*)hevent_userdata(io);
+        if (cli)
         {
-            long remaining = (timeout-time);
-            if(remaining < TIMEOUT_TIMEUNIT) {
-            } else {
+            cli->onClosed();
+        } 
+    });
+    hio_read(cli_io);
 
-            }
-        } else {
-            auto serverProxy = UdpConnMap<uint32_t, UdpServerBoltProxy>::getInstance().get(session_id);
-            if (!serverProxy)
+    timer = htimer_add(loop, [](htimer_t *timer) {
+        UdpClientProxy* cli = (UdpClientProxy*)hevent_userdata(timer);
+        if (cli)
+        {
+            long timeout = cli->isDns() ? DNS_TIMEOUT : TIMEOUT_TIMEUNIT;
+            long time = currentTimeMillis() - cli->ts;
+            if (time > 0 && time < timeout)
             {
-                close();
-                return;
+                long remaining = (timeout - time);
+                if (remaining < TIMEOUT_TIMEUNIT)
+                {
+                    htimer_reset(timer, remaining);
+                }
+                else
+                {
+                    htimer_reset(timer, TIMEOUT_TIMEUNIT);
+                }
             }
-            serverProxy->closeClient(t_addr, u_addr);
-        }
-        
-    });
+            else
+            {
+                cli->close();
+            } 
+        } 
+    }, TIMEOUT_TIMEUNIT);
+    hevent_set_userdata(timer, this);
+
     return true;
 }
 
 void UdpClientProxy::close()
 {
-    cli.closesocket();
-    timer.stop();
+    if (timer)
+    {
+        htimer_del(timer);
+        timer = nullptr;
+    }
+    if (cli_io) {
+        hio_close(cli_io);
+        cli_io = nullptr;
+    }
 }
 
-void UdpClientProxy::onRecv(const hv::SocketChannelPtr &channel, hv::Buffer *buf)
+void UdpClientProxy::onRecv(void *buf, int readbytes)
 {
-    spdlog::info("<<< size:{} : data: {}", (int)buf->size(), (char *)buf->data());
+    spdlog::info("UdpClientProxy::onRecv {}:{}<==>{}:{} <<< size:{} : data: {}", inet_ntoa(u_addr.sin_addr), ntohs(u_addr.sin_port), inet_ntoa(t_addr.sin_addr), ntohs(t_addr.sin_port), readbytes, std::string((char *)buf, readbytes).c_str());
     auto serverProxy = UdpConnMap<uint32_t, UdpServerBoltProxy>::getInstance().get(session_id);
     if (!serverProxy)
     {
         return;
     }
     ts = currentTimeMillis();
-    serverProxy->sendData(t_addr, u_addr, buf->data(), buf->size());
+    serverProxy->sendData(t_addr, u_addr, buf, readbytes);
 }
 
-void UdpClientProxy::onWrited(const hv::SocketChannelPtr &channel, hv::Buffer *buf)
+void UdpClientProxy::onWrited(const void *buf, int writebytes)
 {
-    spdlog::info(">>> size:{} : data: {}", (int)buf->size(), (char *)buf->data());
+    spdlog::info("UdpClientProxy::onWrited {}:{}<==>{}:{} >>> size:{} : data: {}", inet_ntoa(u_addr.sin_addr), ntohs(u_addr.sin_port), inet_ntoa(t_addr.sin_addr), ntohs(t_addr.sin_port), writebytes, std::string((char *)buf, writebytes).c_str());
+}
+
+void UdpClientProxy::onClosed()
+{
+    spdlog::info("UdpClientProxy::onClosed {}:{}<==>{}:{}", inet_ntoa(u_addr.sin_addr), ntohs(u_addr.sin_port), inet_ntoa(t_addr.sin_addr), ntohs(t_addr.sin_port));
+    auto key = get_key(t_addr, u_addr);
+    auto serverProxy = UdpConnMap<uint32_t, UdpServerBoltProxy>::getInstance().get(session_id);
+    if (!serverProxy)
+    {
+        return;
+    }
+    serverProxy->closeClient(t_addr, u_addr);
 }

src/udp_inbound.cpp

@@ -4,6 +4,7 @@
 #include "utils.h"
 #include "bolt/datagram.h"
 #include "bolt/crypt.h"
+#include "spdlog/spdlog.h"
 
 void udp_on_recvfrom(hio_t *io, void *buf, int readbytes)
 {
@@ -35,7 +36,7 @@ void udp_on_recvfrom(hio_t *io, void *buf, int readbytes)
 
                     GENERATE_DECRYPT_KEY(extend_response, extend_response_len, serverProxy->getConfig().ept_type, serverProxy->getConfig().ept_key)
 
-                    char result[1] = {BOLT_BIND_RESPONSE_CODE_SUCESS};
+                    uint32_t result = BOLT_BIND_RESPONSE_CODE_SUCESS;
 
                     PACK_BIND_RESPONSE_DATA(bind_response, bind_response_len, BOLT_CHANNEL_CMD_BIND_RESPONSE, request_id, session_id, result)
 
@@ -44,6 +45,7 @@ void udp_on_recvfrom(hio_t *io, void *buf, int readbytes)
 
                     auto proxy = std::unique_ptr<UdpServerBoltProxy>(serverProxy);
                     UdpConnMap<uint32_t, UdpServerBoltProxy>::getInstance().add(session_id, proxy);
+                    spdlog::info("[UDP]BOLT_CHANNEL_CMD_BIND_REQUEST: requestId:{}, session_id:{}, encrypt:{}", request_id, session_id, serverProxy->getConfig().encrypt);
                 }
                 else if (dest_len == 9 && dest[0] == BOLT_CHANNEL_CMD_UNBIND_REQUEST)
                 {
@@ -51,6 +53,7 @@ void udp_on_recvfrom(hio_t *io, void *buf, int readbytes)
 
                     auto serverProxy = UdpConnMap<uint32_t, UdpServerBoltProxy>::getInstance().get(session_id);
                     serverProxy->recycle();
+                    spdlog::info("[UDP]BOLT_CHANNEL_CMD_UNBIND_REQUEST: requestId:{}, session_id:{}", serverProxy->getConfig().request_id, serverProxy->getConfig().session_id);
                     UdpConnMap<uint32_t, UdpServerBoltProxy>::getInstance().remove(session_id);
                 }
                 else