a8/a8/tcpclient2.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/tcpclient2.h>

const int MAX_RECV_BUFFERSIZE = 1024 * 10;

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

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

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

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

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

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

    void TcpClient2::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 TcpClient2::SetActive(bool active)
    {
        if (active) {
            if (IsActive()) {
                return;
            }
        } else {
          ActiveStop();
        }
    }

    bool TcpClient2::ActiveStart()
    {
        socket_ = ::socket(AF_INET, SOCK_STREAM, IPPROTO_TCP);
        if (INVALID_SOCKET == socket_) {
            if (on_error) {
                on_error(this, errno);
            }
            return false;
        }
        //set nodelay
        {
          int flag = 1;
          int ret = ::setsockopt(socket_,
                                 IPPROTO_TCP,
                                 TCP_NODELAY,
                                 (char *)&flag,
                                 sizeof(flag));
          assert(ret >= 0);
          if (ret < 0) {
            abort();
          }
        }
        //set nonblock
        {
          int flags = 0;
          flags = ::fcntl(socket_, F_GETFL, 0);
          ::fcntl(socket_, F_SETFL, flags|O_NONBLOCK);
        }
        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 TcpClient2::ActiveStop()
    {
        connected_ = false;
        if (socket_ != INVALID_SOCKET) {
            shutdown(socket_, 2);
            ::close(socket_);
        }
        socket_ = INVALID_SOCKET;
    }

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

}