tcp_inbound.cpp

#include "include/tcp_inbound.h"
#include "include/conn_map.h"
#include "include/accumulator.h"
#include "hv/hsocket.h"
#include "hv/hthread.h"
#include "hv/TcpClient.h"
#include "spdlog/spdlog.h"
#include <string>

class TcpClientShell {
    public:
        bool init(hio_t* _io) {
            io = _io;
            int connfd = cli.createsocket(hio_peeraddr(io));
            if (connfd < 0) {
                return false;
            }
            cli.onConnection = [this](const hv::SocketChannelPtr& channel) {
                std::string peeraddr = channel->peeraddr();
                if (channel->isConnected()) {
                    spdlog::info("connected to {}! connfd={}\n", peeraddr.c_str(), channel->fd());
                    if (wait_send_buf.getDataSize() > 0)
                    {
                        cli.send(wait_send_buf.getData(), wait_send_buf.getDataSize());
                    }
                } else {
                    spdlog::info("disconnected to {}! connfd={}\n", peeraddr.c_str(), channel->fd());
                    hio_close(io);
                }
            };

            cli.onMessage = [this](const hv::SocketChannelPtr& channel, hv::Buffer* buf) {
                hio_write(io, buf->data(), buf->size());
                spdlog::info("< %.*s\n", (int)buf->size(), (char*)buf->data());
            };

            cli.onWriteComplete = [this](const hv::SocketChannelPtr& channel, hv::Buffer* buf) {

            };
            cli.start();
            return true;
        }

        int send(const char* data, int size) {
            if (cli.isConnected())
            {
                return cli.send(data, size);
            } else {
                wait_send_buf.addData(data, size);
                return size;
            }
        }

        void close() {
            cli.closesocket();
        }

    private:
        hv::TcpClient cli;
        hio_t* io;
        Accumulator wait_send_buf;
};

static void tcp_on_close(hio_t* io) {
    spdlog::info("tcp_on_close fd={} error={}\n", hio_fd(io), hio_error(io));
    ConnMap<hio_t*, TcpClientShell>::getInstance().remove(io);
}

static void tcp_on_recv(hio_t* io, void* buf, int readbytes) {
    spdlog::info("tcp_on_recv fd={} buf({})={}\n", hio_fd(io), readbytes, (const char*)buf);
    hio_write(io, buf, readbytes);
    auto cli = ConnMap<hio_t*, TcpClientShell>::getInstance().get(io);
    if(cli) {
        cli->send((const char*) buf, readbytes);
    }
}

void tcp_on_accept(hio_t* io, hevent_t* ev) {
    hloop_t* loop = ev->loop;

    char localaddrstr[SOCKADDR_STRLEN] = {0};
    char peeraddrstr[SOCKADDR_STRLEN] = {0};
    spdlog::info("tcp_on_accept tid={} connfd={} [{}] <= [{}]\n",
            (long)hv_gettid(),
            (int)hio_fd(io),
            SOCKADDR_STR(hio_localaddr(io), localaddrstr),
            SOCKADDR_STR(hio_peeraddr(io), peeraddrstr));

    hio_setcb_close(io, tcp_on_close);
    hio_setcb_read(io, tcp_on_recv);
    hio_read(io);
}