src/FeedHandlerListener.cpp

#include "FeedHandlerListener.h"

FeedHandlerListener::FeedHandlerListener()
  : mConnId(0)
  //, mSPSC(4096)
{}

FeedHandlerListener::~FeedHandlerListener()
{
    auto [id, err] = close();

    if(id < 0 && err != "") std::cout << err << '\n';
}

void FeedHandlerListener::setConnData(const WebSockConnData& data)
{
    mConnData.host = std::move(data.host);
    mConnData.port = std::move(data.port);
    mConnData.endpoint = std::move(data.endpoint);
    mConnData.instruments = std::move(data.instruments);
    mConnData.priceMask = std::move(data.priceMask);
    mConnData.tickSize = std::move(data.tickSize);
    mConnData.slidingWindowSize = std::move(data.slidingWindowSize);
}

std::pair<int, std::string> FeedHandlerListener::connect()
{
    auto [id, err] = mWebSock.connect(mConnData.host, mConnData.port, mConnData.endpoint, mConnData.instruments);

    if (id <= 0)
    {
        std::string msg = "The WebSocket connection failed. ['" + mConnData.toStr() + "'] Reason: " + err;
        return {-1, msg};
    }

    mConnId = id;
    boost::split(mInstrumentBuffer, mConnData.instruments, boost::is_any_of(","));

    mSlidingWindowBuffer.reserve(mInstrumentBuffer.size());
    mCSVs.reserve(mInstrumentBuffer.size());

    std::vector<std::string> instrPxMaskList;
    boost::split(instrPxMaskList, mConnData.priceMask, boost::is_any_of(","));
    std::vector<std::string> instrTickSizeList;
    boost::split(instrTickSizeList, mConnData.tickSize, boost::is_any_of(","));

    for(size_t i = 0; i < mInstrumentBuffer.size(); i++)
        mSlidingWindowBuffer.emplace_back(std::stoll(mConnData.slidingWindowSize), 2048, 16384, std::stoll(instrPxMaskList[i]) * 2, std::stoll(instrTickSizeList[i]));

    for(size_t i = 0; i < mInstrumentBuffer.size(); i++)
        mCSVs.emplace_back("CSV/" + mInstrumentBuffer[i] + ".csv");

    connectionHandler();
    return {0, err};
}

std::pair<int, std::string> FeedHandlerListener::close()
{
    mProdThrRunning.store(false, std::memory_order_relaxed);
    mConsThrRunning.store(false, std::memory_order_relaxed);

    if(!mConnId) return {-1, ""};

    auto [id, err] = mWebSock.close(mConnId);

    if(id <= 0)
    {
        std::string msg = "Error during websocket destructor cleanup: '" + err + '\'';
        return {-1, msg};
    }

    return {0, ""};
}

void FeedHandlerListener::wait()
{
    if(mProdThr.joinable())
    {
        mProdThr.join();
        std::cout << "Producer stopped\n";
    }

    if(mConsThr.joinable())
    {
        mConsThr.join();
        std::cout << "Consumer stopped\n";
    }
}

void FeedHandlerListener::connectionHandler()
{
    mProdThrRunning.store(true, std::memory_order_relaxed);
    mConsThrRunning.store(true, std::memory_order_relaxed);

    mProdThr = std::thread(&FeedHandlerListener::producer, this);
    mConsThr = std::thread(&FeedHandlerListener::consumer, this);
}

void FeedHandlerListener::subscribe()
{
    std::string subInstr = "";
    
    for(size_t i = 0; i < mInstrumentBuffer.size(); i++)
        subInstr += '"' + mInstrumentBuffer[i] + '"' + (i + 1 != mInstrumentBuffer.size() ? "," : "");

    std::string json = R"({"type":"subscribe","product_ids":[)" + subInstr + R"(],"channels":["ticker"]})";

    auto [id, err] = mWebSock.send(mConnId, json);
    if(id <= 0)
    {
        std::string e = "Failed to send subscription msg: [" + err + ']';
        std:: cout << e << '\n';
    }

    std::cout << json << '\n';
    std::cout << "Subscription sent. [" + subInstr + "]\n";
}

void FeedHandlerListener::producer()
{
    std::cout << "Producer started\n";

    subscribe();

    simdjson::dom::parser parser;
    TickEntryParser tickParser(mInstrumentBuffer);

    while(mProdThrRunning.load(std::memory_order_relaxed))
    {
        auto [recvStatus, recvBuffer] = mWebSock.recv(mConnId);

        if(recvBuffer.find("error") != std::string::npos)
        {
            std::string e = "JSON error: [" + recvBuffer + ']';
            std:: cout << e << '\n';
        }

        if(recvStatus > 0) //Critical path
        {
            simdjson::dom::element ticker;

            if(parser.parse(recvBuffer).get(ticker) != simdjson::SUCCESS)
            {
                std::cout << "PRODUCER: Error parsing to JSON. [" + recvBuffer + "]\n";
            }
            else
            {
                //std::cout << ticker << '\n';
                auto [status, tick] = tickParser.parse(ticker);
                if(status) if(!mSPSC.push(std::move(tick))) 
                    std::cout << "Producer ERROR: SPSC queue is full\n";
            }
        }
        else
        {
            std::cout << "Websocket connectivity dropped...\n";
        }
    }
}

void FeedHandlerListener::consumer()
{
    std::cout << "Consumer started\n";

    while(mConsThrRunning.load(std::memory_order_relaxed))
    {
        TickEntry tick;
        if(mSPSC.pop(tick))
        {
            auto& slidingWindow = mSlidingWindowBuffer[tick.mSymIdx];
            auto [status, data] = slidingWindow.push(tick);
            std::string csv = "";
            if(status)
                csv = tick.toStr() + ',' + data.toStr();
            else
                csv = "ERROR: Cannot push to sliding window, buffer is full. [" + tick.toStr() + "]";

            std::cout << csv << " - " << slidingWindow.getSize() << " - " << mInstrumentBuffer[tick.mSymIdx] << '\n';
            //assert(data.mMedian >= data.mLow && data.mMedian <= data.mHigh);

            if(!mCSVs[tick.mSymIdx].push(tick, data))
                std::cout << "ERROR: Cannot push to CSV file\n";
        }	
    }
}