a8/a8/asiotcpclient.cc

#include <string.h>
#include <assert.h>

#include <mutex>
#include <thread>
#include <iostream>

#include <a8/a8.h>
#include <a8/asiotcpclient.h>

#ifdef USE_BOOST

const int MAX_RECV_BUFFERSIZE = 1024 * 64;

namespace a8
{

    AsioTcpClient::AsioTcpClient(std::shared_ptr<asio::io_context> io_context, const std::string& remote_ip, int remote_port)
    {
        io_context_ = io_context;
        remote_address_ = remote_ip;
        remote_port_ = remote_port;
        endpoint_ = std::make_shared<asio::ip::tcp::endpoint>
            (
             asio::ip::address::from_string(remote_address_),
              remote_port_
             );
        send_buffer_mutex_ = std::make_shared<std::mutex>();
        socket_ = std::make_shared<asio::ip::tcp::socket>(*io_context);
    }

    AsioTcpClient::~AsioTcpClient()
    {
        Close();
    }

    void AsioTcpClient::Open()
    {
        if (!IsActive()) {
            SetActive(true);
        }
    }

    void AsioTcpClient::Close()
    {
        if (IsActive()) {
            SetActive(false);
        }
    }

    bool AsioTcpClient::IsActive()
    {
        return actived_;
    }

    bool AsioTcpClient::Connected()
    {
        return connected_;
    }

    void AsioTcpClient::SendBuff(const char* buff, unsigned int bufflen)
    {
        //a8::XPrintf("SendBuff bufflen:%d\n", {bufflen});
        if (bufflen > 0) {
            a8::SendQueueNode* p = (a8::SendQueueNode*)malloc(sizeof(a8::SendQueueNode));
            memset(p, 0, sizeof(SendQueueNode));
            p->buff = (char*)malloc(bufflen);
            memmove(p->buff, buff, bufflen);
            p->bufflen = bufflen;
            send_buffer_mutex_->lock();
            if (bot_node_) {
                bot_node_->next = p;
                bot_node_ = p;
            }else{
                top_node_ = p;
                bot_node_ = p;
            }
            send_buffer_mutex_->unlock();
            DoSend();
        }
    }

    void AsioTcpClient::SetActive(bool active)
    {
        if (active) {
            if (!IsActive()) {
                ActiveStart();
            }
        } else {
            if (IsActive()) {
                ActiveStop();
            }
        }
    }

    void AsioTcpClient::ActiveStart()
    {
        actived_ = true;
        connected_ = false;
        socket_->async_connect
            (*endpoint_,
             [this] (const asio::error_code& ec)
             {
                 HandleConnect(ec);
             });
    }

    void AsioTcpClient::ActiveStop()
    {
        actived_ = true;
        connected_ = false;
    }

    void AsioTcpClient::HandleConnect(const asio::error_code& err)
    {
        if (err) {
            actived_ = false;
            connected_ = false;
            if (on_error) {
                on_error(this, err.value());
            }
            return;
        } else {
            connected_ = true;
            if (on_connect) {
                on_connect(this);
            }
            DoRead();
        }
    }

    void AsioTcpClient::DoRead()
    {
        socket_->async_read_some
            (asio::buffer(buffer_),
             [this](std::error_code ec, std::size_t bytes_transferred)
             {
                 if (!ec) {
                     if (on_socketread) {
                         on_socketread(this, buffer_.data(), bytes_transferred);
                     }
                     DoRead();
                 } else {
                     a8::XPrintf("DoRead error %s\n", {ec.message()});
                     actived_ = false;
                     connected_ = false;
                     if (on_disconnect) {
                         on_disconnect(this, ec.value());
                     }
                 }
             });
    }

    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_ && !sending_) {
            sending_ = true;
            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) {
                         send_buffer_mutex_->lock();
                         if (work_node_) {
                             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);
                             }
                             if (!work_node_) {
                                 sending_ = false;
                             }
                             send_buffer_mutex_->unlock();
                             DoSend();
                             return;
                         }
                         send_buffer_mutex_->unlock();
                     } else {
                         a8::XPrintf("DoSend error %s\n", {ec.message()});
                         actived_ = false;
                         connected_ = false;
                         if (on_disconnect) {
                             on_disconnect(this, ec.value());
                         }
                     }
                 });

        }
    }

}

#endif