stream_inlet.cpp

Go to the documentation of this file.

//=============================================================================================================
 
//=============================================================================================================
// INCLUDES
//=============================================================================================================
 
#include "stream_inlet.h"
 
//=============================================================================================================
// QT INCLUDES
//=============================================================================================================
 
#include <QTcpSocket>
#include <QByteArray>
#include <QDebug>
 
//=============================================================================================================
// STL INCLUDES
//=============================================================================================================
 
#include <cstring>
#include <stdexcept>
 
//=============================================================================================================
// USED NAMESPACES
//=============================================================================================================
 
using namespace LSLLIB;
 
//=============================================================================================================
// PRIVATE IMPLEMENTATION
//=============================================================================================================

class LSLLIB::StreamInletPrivate
{
public:
    StreamInletPrivate(const stream_info& info)
    : m_info(info)
    , m_pSocket(nullptr)
    , m_bIsOpen(false)
    , m_iChannelCount(info.channel_count())
    , m_iBytesPerSample(static_cast<int>(info.channel_count() * sizeof(float)))
    {
    }
 
    ~StreamInletPrivate()
    {
        closeStream();
    }
 
    //=========================================================================================================
    void openStream()
    {
        if (m_bIsOpen) {
            return;
        }
 
        // Create socket (no parent, we manage lifetime ourselves)
        m_pSocket = new QTcpSocket();
 
        QString host = QString::fromStdString(m_info.data_host());
        quint16 port = static_cast<quint16>(m_info.data_port());
 
        if (host.isEmpty() || port == 0) {
            delete m_pSocket;
            m_pSocket = nullptr;
            throw std::runtime_error("[lsl::stream_inlet] Invalid data host or port in stream_info");
        }
 
        m_pSocket->connectToHost(host, port);
 
        if (!m_pSocket->waitForConnected(5000)) {
            QString err = m_pSocket->errorString();
            delete m_pSocket;
            m_pSocket = nullptr;
            throw std::runtime_error(std::string("[lsl::stream_inlet] Failed to connect to outlet: ")
                                     + err.toStdString());
        }
 
        // Read the handshake header: "LSL1" (4 bytes) + channel_count (4 bytes, little-endian)
        if (!m_pSocket->waitForReadyRead(5000)) {
            delete m_pSocket;
            m_pSocket = nullptr;
            throw std::runtime_error("[lsl::stream_inlet] Timeout waiting for handshake from outlet");
        }
 
        QByteArray header;
        while (header.size() < 8) {
            if (m_pSocket->bytesAvailable() == 0) {
                if (!m_pSocket->waitForReadyRead(5000)) {
                    break;
                }
            }
            header.append(m_pSocket->read(8 - header.size()));
        }
 
        if (header.size() < 8 || header.left(4) != QByteArray("LSL1", 4)) {
            delete m_pSocket;
            m_pSocket = nullptr;
            throw std::runtime_error("[lsl::stream_inlet] Invalid handshake from outlet");
        }
 
        // Read channel count from header (little-endian int32)
        int headerChannels = 0;
        std::memcpy(&headerChannels, header.constData() + 4, sizeof(int));
        if (headerChannels != m_iChannelCount) {
            qDebug() << "[lsl::stream_inlet] Warning: outlet reports" << headerChannels
                     << "channels, expected" << m_iChannelCount << "- using outlet value";
            m_iChannelCount = headerChannels;
            m_iBytesPerSample = m_iChannelCount * static_cast<int>(sizeof(float));
        }
 
        m_bIsOpen = true;
    }
 
    //=========================================================================================================
    void closeStream()
    {
        if (m_pSocket) {
            if (m_pSocket->state() == QAbstractSocket::ConnectedState) {
                m_pSocket->disconnectFromHost();
                if (m_pSocket->state() != QAbstractSocket::UnconnectedState) {
                    m_pSocket->waitForDisconnected(1000);
                }
            }
            delete m_pSocket;
            m_pSocket = nullptr;
        }
        m_bIsOpen = false;
        m_rawBuffer.clear();
    }
 
    //=========================================================================================================
    bool readPending()
    {
        if (!m_bIsOpen || !m_pSocket) {
            return false;
        }
 
        // Non-blocking check for available data
        if (m_pSocket->bytesAvailable() > 0 || m_pSocket->waitForReadyRead(0)) {
            QByteArray data = m_pSocket->readAll();
            m_rawBuffer.append(data);
        }
 
        // A complete sample requires m_iBytesPerSample bytes
        return (m_iBytesPerSample > 0) && (m_rawBuffer.size() >= m_iBytesPerSample);
    }
 
    //=========================================================================================================
    std::vector<std::vector<float>> pullChunkFloat()
    {
        std::vector<std::vector<float>> chunk;
 
        if (!m_bIsOpen || !m_pSocket) {
            return chunk;
        }
 
        // Read any pending data
        readPending();
 
        if (m_iBytesPerSample <= 0) {
            return chunk;
        }
 
        // Extract complete samples from the buffer
        int nCompleteSamples = m_rawBuffer.size() / m_iBytesPerSample;
        if (nCompleteSamples == 0) {
            return chunk;
        }
 
        chunk.reserve(nCompleteSamples);
        const char* ptr = m_rawBuffer.constData();
 
        for (int s = 0; s < nCompleteSamples; ++s) {
            std::vector<float> sample(m_iChannelCount);
            std::memcpy(sample.data(), ptr + s * m_iBytesPerSample, m_iBytesPerSample);
            chunk.push_back(std::move(sample));
        }
 
        // Remove consumed bytes from the buffer
        int consumedBytes = nCompleteSamples * m_iBytesPerSample;
        m_rawBuffer.remove(0, consumedBytes);
 
        return chunk;
    }
 
    stream_info     m_info;             
    QTcpSocket*     m_pSocket;          
    bool            m_bIsOpen;          
    int             m_iChannelCount;    
    int             m_iBytesPerSample;  
    QByteArray      m_rawBuffer;        
};
 
//=============================================================================================================
// DEFINE MEMBER METHODS
//=============================================================================================================
 
stream_inlet::stream_inlet(const stream_info& info)
: m_pImpl(new StreamInletPrivate(info))
{
}
 
//=============================================================================================================
 
stream_inlet::~stream_inlet()
{
    // unique_ptr destructor handles cleanup via StreamInletPrivate destructor
}
 
//=============================================================================================================
 
void stream_inlet::open_stream()
{
    m_pImpl->openStream();
}
 
//=============================================================================================================
 
void stream_inlet::close_stream()
{
    m_pImpl->closeStream();
}
 
//=============================================================================================================
 
bool stream_inlet::samples_available()
{
    return m_pImpl->readPending();
}
 
//=============================================================================================================
 
std::vector<std::vector<float>> stream_inlet::pull_chunk_float()
{
    return m_pImpl->pullChunkFloat();
}