1

a8/asiotcpclient.cc

@@ -2,17 +2,8 @@
 #include <assert.h>
 
 #include <mutex>
-#include <condition_variable>
 #include <thread>
 
-#include <unistd.h>
-#include <arpa/inet.h>
-#include <fcntl.h>
-#include <sys/epoll.h>
-#include <netinet/tcp.h>
-
-#include <boost/bind.hpp>
-
 #include <a8/a8.h>
 #include <a8/asiotcpclient.h>
 
@@ -26,6 +17,10 @@ namespace a8
     AsioTcpClient::AsioTcpClient()
     {
         send_buffer_mutex_ = std::make_shared<std::mutex>();
+        io_context_ = std::make_shared<asio::io_context>();
+        resolver_ = std::make_shared<asio::ip::tcp::resolver>(*io_context_);
+        socket_ = std::make_shared<asio::ip::tcp::socket>(*io_context_);
+        new std::thread(&AsioTcpClient::WorkerThreadProc, this);
     }
 
     AsioTcpClient::~AsioTcpClient()
@@ -74,6 +69,7 @@ namespace a8
                 bot_node_ = p;
             }
             send_buffer_mutex_->unlock();
+            DoSend();
         }
     }
 
@@ -152,6 +148,48 @@ namespace a8
              });
     }
 
+    void AsioTcpClient::DoSend()
+    {
+        if (!work_node_) {
+            send_buffer_mutex_->lock();
+            work_node_ = top_node_;
+            top_node_ = nullptr;
+            bot_node_ = nullptr;
+            send_buffer_mutex_->unlock();
+        }
+        if (work_node_) {
+            char* buf = work_node_->buff + work_node_->sent_bytes;
+            int buf_len = work_node_->bufflen - work_node_->sent_bytes;
+            asio::async_write
+                (*socket_,
+                 asio::buffer(buf, buf_len),
+                 [this] (const asio::error_code& ec, std::size_t bytes_transferred)
+                 {
+                     if (!ec) {
+                         work_node_->sent_bytes += bytes_transferred;
+                         if (work_node_->sent_bytes >= work_node_->bufflen) {
+                             auto pdelnode = work_node_;
+                             work_node_ = work_node_->next;
+                             free(pdelnode->buff);
+                             free((void*)pdelnode);
+                         }
+                         DoSend();
+                     } else {
+                         abort();
+                     }
+                 });
+
+        }
+    }
+
+    void AsioTcpClient::WorkerThreadProc()
+    {
+        while (true) {
+            io_context_->run();
+            int i = 0;
+        }
+    }
+
 }
 
 #endif

a8/asiotcpclient.h

@@ -31,8 +31,11 @@ namespace a8
     private:
         void HandleConnect(const asio::error_code& err, const tcp::endpoint& endpoint);
         void DoRead();
+        void DoSend();
+        void WorkerThreadProc();
 
     private:
+        std::shared_ptr<asio::io_context> io_context_;
         std::shared_ptr<asio::ip::tcp::resolver> resolver_;
         std::shared_ptr<asio::ip::tcp::socket> socket_;
         volatile bool actived_ = false;
@@ -40,6 +43,7 @@ namespace a8
         std::shared_ptr<std::mutex> send_buffer_mutex_;
         SendQueueNode *top_node_ = nullptr;
         SendQueueNode *bot_node_ = nullptr;
+        volatile SendQueueNode *work_node_ = nullptr;
         std::array<char, 1024 * 64> buffer_;
 
         void SetActive(bool active);