a8/a8/tcpclient.cc

#include <string.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 <a8/a8.h>
#include <a8/tcpclient.h>

const int MAX_RECV_BUFFERSIZE = 1024 * 10;

namespace a8
{
    TcpClient::TcpClient()
    {
        send_buffer_mutex_ = new std::mutex();
        send_cond_mutex_ = new std::mutex();
        send_cond_ = new std::condition_variable();
    }

    TcpClient::~TcpClient()
    {
        Close();
        delete send_buffer_mutex_;
        send_buffer_mutex_ = nullptr;
        delete send_cond_mutex_;
        send_cond_mutex_ = nullptr;
        delete send_cond_;
        send_cond_ = nullptr;
    }

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

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

    bool TcpClient::IsActive()
    {
        return socket_ != a8::INVALID_SOCKET;
    }

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

    void TcpClient::SendBuff(const char* buff, unsigned int 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();
            NotifySendCond();
        }
    }

    void TcpClient::SetActive(bool active)
    {
        if (active) {
            if (IsActive()) {
                return;
            }
            if (worker_thread_) {
                ActiveStop();
            }
            worker_thread_shutdown_ = false;
            worker_thread_ = new std::thread(&TcpClient::WorkerThreadProc, this);
        } else {
            if (worker_thread_) {
                ActiveStop();
            }
        }
    }

    bool TcpClient::ActiveStart()
    {
        socket_ = ::socket(AF_INET, SOCK_STREAM, IPPROTO_TCP);
        if (INVALID_SOCKET == socket_) {
            if (on_error) {
                on_error(this, errno);
            }
            return false;
        }
        sockaddr_in sa;
        memset(&sa, 0, sizeof(sa));
        sa.sin_family = AF_INET;
        sa.sin_addr.s_addr = inet_addr(remote_address.c_str());
        sa.sin_port = htons(remote_port);
        if (::connect(socket_, (sockaddr*)&sa, sizeof(sa)) < 0) {
            if (on_error) {
                on_error(this, errno);
            }
            ::close(socket_);
            socket_ = INVALID_SOCKET;
            return false;
        }
        connected_ = true;
        if (on_connect) {
            on_connect(this);
        }
        return true;
    }

    void TcpClient::ActiveStop()
    {
        connected_ = false;
        if (socket_ != INVALID_SOCKET) {
            shutdown(socket_, 2);
            ::close(socket_);
        }
        if (worker_thread_) {
            worker_thread_shutdown_ = true;
            worker_thread_->join();
            delete worker_thread_;
            worker_thread_ = nullptr;
        }
        socket_ = INVALID_SOCKET;
    }

    void TcpClient::WorkerThreadProc()
    {
        if (!ActiveStart()) {
            return;
        }

        sender_thread_shutdown_ = false;
        std::thread* senderthread = new std::thread(&TcpClient::SenderThreadProc, this);
        char recvBuf[MAX_RECV_BUFFERSIZE];
        while (!worker_thread_shutdown_) {
            int ret = ::recv(socket_, recvBuf, MAX_RECV_BUFFERSIZE, 0);
            if (ret < 0) {
                connected_ = false;
                if (on_disconnect) {
                    on_disconnect(this);
                }
                worker_thread_shutdown_ = true;
                break;
            } else if(ret == 0) {
                connected_ = false;
                if (on_disconnect) {
                    on_disconnect(this);
                }
                worker_thread_shutdown_ = true;
                break;
            } else {
                if (on_socketread) {
                    on_socketread(this, recvBuf, (unsigned int)ret);
                }
            }
        }
        sender_thread_shutdown_ = true;
        senderthread->join();
        delete senderthread;
        senderthread = nullptr;
        socket_ = INVALID_SOCKET;
    }

    void TcpClient::SenderThreadProc()
    {
        a8::SendQueueNode* worknode = nullptr;
        a8::SendQueueNode* currnode = nullptr;
        while (!sender_thread_shutdown_) {
            if (!worknode && top_node_) {
                send_buffer_mutex_->lock();
                worknode = top_node_;
                top_node_ = nullptr;
                bot_node_ = nullptr;
                send_buffer_mutex_->unlock();
            }

            while (worknode && !sender_thread_shutdown_) {
                currnode = worknode;
                while (currnode->sent_bytes < currnode->bufflen && !sender_thread_shutdown_) {
                    int len = ::send(socket_,
                                     currnode->buff + currnode->sent_bytes,
                                     currnode->bufflen - currnode->sent_bytes,
                                     0);
                    if (len > 0) {
                        currnode->sent_bytes += len;
                    } else {
                        break;
                    }
                }
                //send
                if (currnode->sent_bytes >= currnode->bufflen) {
                    worknode = worknode->next;
                    free(currnode->buff);
                    free(currnode);
                }

            }
            {
                std::unique_lock<std::mutex> lk(*send_cond_mutex_);
                send_cond_->wait_for(lk, std::chrono::seconds(1));
            }
        }
        while (worknode) {
            currnode = worknode;
            worknode = worknode->next;
            free(currnode->buff);
            free(currnode);
        }
    }

    void TcpClient::NotifySendCond()
    {
        std::unique_lock<std::mutex> lk(*send_cond_mutex_);
        send_cond_->notify_all();
    }

}