jbus/lib/Listener.cpp

#include "jbus/Listener.hpp"
#include "jbus/Endpoint.hpp"

#define LOG_LISTENER 0

namespace jbus
{

void Listener::listenerProc()
{
#if LOG_LISTENER
    printf("JoyBus listener started\n");
#endif

    net::IPAddress localhost("127.0.0.1");
    bool dataBound = false;
    bool clockBound = false;
    while (m_running && (!dataBound || !clockBound))
    {
        if (!dataBound)
        {
            if (!(dataBound = m_dataServer.openAndListen(localhost, DataPort)))
            {
                m_dataServer = net::Socket(false);
#if LOG_LISTENER
                printf("data open failed %s; will retry\n", strerror(errno));
#endif
                WaitGCTicks(GetGCTicksPerSec());
            }
            else
            {
#if LOG_LISTENER
                printf("data listening on port %d\n", DataPort);
#endif
            }
        }
        if (!clockBound)
        {
            if (!(clockBound = m_clockServer.openAndListen(localhost, ClockPort)))
            {
                m_clockServer = net::Socket(false);
#if LOG_LISTENER
                printf("clock open failed %s; will retry\n", strerror(errno));
#endif
                WaitGCTicks(GetGCTicksPerSec());
            }
            else
            {
#if LOG_LISTENER
                printf("clock listening on port %d\n", ClockPort);
#endif
            }
        }
    }

    /* We use blocking I/O since we have a dedicated transfer thread */
    net::Socket acceptData = {true};
    net::Socket acceptClock = {true};
    std::string hostname;
    while (m_running)
    {
        if (m_dataServer.accept(acceptData, hostname) == net::Socket::EResult::OK)
        {
#if LOG_LISTENER
            printf("accepted data connection from %s\n", hostname.c_str());
#endif
        }
        if (m_clockServer.accept(acceptClock, hostname) == net::Socket::EResult::OK)
        {
#if LOG_LISTENER
            printf("accepted clock connection from %s\n", hostname.c_str());
#endif
        }
        if (acceptData && acceptClock)
        {
            std::unique_lock<std::mutex> lk(m_queueLock);
            m_endpointQueue.push(std::make_unique<Endpoint>(
                0, std::move(acceptData), std::move(acceptClock)));
        }
        WaitGCTicks(GetGCTicksPerSec());
    }

    m_dataServer.close();
    m_clockServer.close();
#if LOG_LISTENER
    printf("JoyBus listener stopped\n");
#endif
}

void Listener::start()
{
    stop();
    m_running = true;
    m_listenerThread = std::thread(std::bind(&Listener::listenerProc, this));
}

void Listener::stop()
{
    m_running = false;
    if (m_listenerThread.joinable())
        m_listenerThread.join();
}

std::unique_ptr<Endpoint> Listener::accept()
{
    std::unique_lock<std::mutex> lk(m_queueLock);
    if (m_endpointQueue.size())
    {
        std::unique_ptr<Endpoint> ret;
        ret = std::move(m_endpointQueue.front());
        m_endpointQueue.pop();
        return ret;
    }
    return {};
}

Listener::~Listener() { stop(); }

}