client.cc

#include "client.h"
#include <vector>
#include <cstdint>

bool Client::IsAlive()
{
    return alive.load(std::memory_order_acquire);
}

void Client::QueueToTcpLoop()
{
    while(alive.load(std::memory_order_acquire) && conn && conn->Valid())
    {

        Frame::EthFrame frame;
        if (!out_queue.Pop(frame)) {
            std::this_thread::sleep_for(std::chrono::milliseconds(1));
            continue;
        }
        conn->Send(frame.data(), frame.size());
    }
}
void Client::PushToOutQueue(const Frame::EthFrame& frame)
{
    out_queue.Push(frame);
}

void Client::TcpToCanLoop(SocketCan* can)
{
    std::vector<uint8_t> inbuf;
    inbuf.reserve(4096);
    uint8_t tmp[2048];

    while (alive.load(std::memory_order_acquire) && conn && conn->Valid())
    {
        ssize_t r = conn->Recv(tmp, sizeof(tmp));
        if (r == 0) break;
        if (r < 0) {
            if (errno == EINTR) continue;
            break;
        }
        inbuf.insert(inbuf.end(), tmp, tmp + r);

        while (inbuf.size() >= PACKED_FRAME_SIZE)
        {
            Frame::EthFrame pf{};
            std::memcpy(pf.data(), inbuf.data(), PACKED_FRAME_SIZE);
            inbuf.erase(inbuf.begin(), inbuf.begin() + PACKED_FRAME_SIZE);

            Frame::CanFrame cf{};
            if (!Frame::Unpack(pf, cf)) continue;
            //(*n_frames)++;
            if (!can->Write(cf)) { alive.store(false, std::memory_order_release); break; }
        }
    }

    alive.store(false, std::memory_order_release);
    if (conn) conn->ShutdownRD();
}