lsl_stream_outlet.cpp

Go to the documentation of this file.

//=============================================================================================================
 
//=============================================================================================================
// INCLUDES
//=============================================================================================================
 
#include "lsl_stream_outlet.h"
 
//=============================================================================================================
// QT INCLUDES
//=============================================================================================================
 
#include <QTcpServer>
#include <QTcpSocket>
#include <QUdpSocket>
#include <QHostAddress>
#include <QDebug>
 
//=============================================================================================================
// STL INCLUDES
//=============================================================================================================
 
#include <thread>
#include <mutex>
#include <atomic>
#include <cstring>
#include <queue>
#include <future>
#include <chrono>
 
//=============================================================================================================
// USED NAMESPACES
//=============================================================================================================
 
using namespace LSLLIB;
 
//=============================================================================================================
// CONSTANTS
//=============================================================================================================
 
namespace
{
const QHostAddress DISCOVERY_MULTICAST_GROUP("239.255.172.215");
const quint16 DISCOVERY_PORT = 16571;
const int BROADCAST_INTERVAL_MS = 500;
}
 
//=============================================================================================================
// PRIVATE IMPLEMENTATION
//=============================================================================================================

class LSLLIB::StreamOutletPrivate
{
public:
    StreamOutletPrivate(const stream_info& info)
    : m_info(info)
    , m_bRunning(false)
    {
    }
 
    ~StreamOutletPrivate()
    {
        stop();
    }
 
    //=========================================================================================================
    void start()
    {
        m_bRunning = true;
        std::future<void> listening = m_listening.get_future();
        m_bgThread = std::thread(&StreamOutletPrivate::run, this);
        listening.wait(); // m_info (with its data port) is not modified after this point
    }
 
    //=========================================================================================================
    void stop()
    {
        m_bRunning = false;
        if (m_bgThread.joinable()) {
            m_bgThread.join();
        }
    }
 
    //=========================================================================================================
    void enqueueSample(const std::vector<float>& sample)
    {
        std::lock_guard<std::mutex> lock(m_queueMutex);
        m_sampleQueue.push(sample);
    }
 
    //=========================================================================================================
    void enqueueChunk(const std::vector<std::vector<float>>& chunk)
    {
        std::lock_guard<std::mutex> lock(m_queueMutex);
        for (const auto& sample : chunk) {
            m_sampleQueue.push(sample);
        }
    }
 
    //=========================================================================================================
    stream_info info() const
    {
        return m_info;
    }
 
    //=========================================================================================================
    bool haveConsumers() const
    {
        return m_nClients.load() > 0;
    }
 
private:
    //=========================================================================================================
    void run()
    {
        // --- Set up TCP server ---
        QTcpServer tcpServer;
        if (!tcpServer.listen(QHostAddress::Any, 0)) {
            qDebug() << "[lsl::stream_outlet] Failed to start TCP server:" << tcpServer.errorString();
            m_listening.set_value();
            return;
        }
 
        // Record the assigned port into stream_info
        m_info.set_data_port(tcpServer.serverPort());
        m_listening.set_value();
 
        // --- Set up UDP socket for multicast discovery ---
        QUdpSocket udpSocket;
#ifndef Q_OS_WASM
        udpSocket.setSocketOption(QAbstractSocket::MulticastTtlOption, 1);
#endif
 
        // --- Prepare the handshake header ---
        // "LSL1" (4 bytes) + channel_count (4 bytes, little-endian int)
        QByteArray handshake("LSL1", 4);
        int ch = m_info.channel_count();
        handshake.append(reinterpret_cast<const char*>(&ch), sizeof(int));
 
        // Prepare discovery datagram (re-created each loop to include up-to-date port)
        std::string discoveryPayload = m_info.to_string();
 
        // Track connected client sockets (owned by this thread)
        std::vector<QTcpSocket*> clients;
 
        auto lastBroadcast = std::chrono::steady_clock::now() - std::chrono::seconds(10); // send immediately
 
        // --- Main loop ---
        while (m_bRunning) {
            // 1. Accept new connections
            while (tcpServer.waitForNewConnection(0)) {
                QTcpSocket* client = tcpServer.nextPendingConnection();
                if (client) {
                    // Send handshake to the new client
                    client->write(handshake);
                    client->flush();
                    clients.push_back(client);
                    m_nClients.store(static_cast<int>(clients.size()));
                }
            }
 
            // 2. Send UDP discovery broadcast (every BROADCAST_INTERVAL_MS)
            auto now = std::chrono::steady_clock::now();
            auto elapsed = std::chrono::duration_cast<std::chrono::milliseconds>(now - lastBroadcast).count();
            if (elapsed >= BROADCAST_INTERVAL_MS) {
                QByteArray datagram(discoveryPayload.c_str(), static_cast<int>(discoveryPayload.size()));
                udpSocket.writeDatagram(datagram, DISCOVERY_MULTICAST_GROUP, DISCOVERY_PORT);
                // Same-host listeners still find the stream when multicast is unrouted (VPNs, CI VMs).
                udpSocket.writeDatagram(datagram, QHostAddress(QHostAddress::LocalHost), DISCOVERY_PORT);
                lastBroadcast = now;
            }
 
            // 3. Drain sample queue and write to all clients
            {
                std::lock_guard<std::mutex> lock(m_queueMutex);
                while (!m_sampleQueue.empty()) {
                    const std::vector<float>& sample = m_sampleQueue.front();
                    QByteArray sampleData(reinterpret_cast<const char*>(sample.data()),
                                          static_cast<int>(sample.size() * sizeof(float)));
 
                    // Write to all clients, remove disconnected ones
                    auto it = clients.begin();
                    while (it != clients.end()) {
                        QTcpSocket* client = *it;
                        if (client->state() != QAbstractSocket::ConnectedState) {
                            delete client;
                            it = clients.erase(it);
                            continue;
                        }
                        client->write(sampleData);
                        ++it;
                    }
 
                    m_sampleQueue.pop();
                }
            }
 
            // Flush all clients
            for (QTcpSocket* client : clients) {
                client->flush();
            }
 
            m_nClients.store(static_cast<int>(clients.size()));
 
            // Sleep briefly to avoid busy-waiting
            std::this_thread::sleep_for(std::chrono::milliseconds(1));
        }
 
        // --- Cleanup ---
        for (QTcpSocket* client : clients) {
            client->disconnectFromHost();
            delete client;
        }
        clients.clear();
        m_nClients.store(0);
 
        tcpServer.close();
        udpSocket.close();
    }
 
    stream_info m_info;             
    std::atomic<bool> m_bRunning;   
    std::thread m_bgThread;         
    std::promise<void> m_listening; 
 
    std::mutex m_queueMutex;                      
    std::queue<std::vector<float>> m_sampleQueue; 
    std::atomic<int> m_nClients{0};               
};
 
//=============================================================================================================
// DEFINE MEMBER METHODS
//=============================================================================================================
 
stream_outlet::stream_outlet(const stream_info& info)
: m_pImpl(new StreamOutletPrivate(info))
{
    m_pImpl->start();
}
 
//=============================================================================================================
 
stream_outlet::~stream_outlet()
{
    // unique_ptr destructor handles cleanup via StreamOutletPrivate destructor
}
 
//=============================================================================================================
 
void stream_outlet::push_sample(const std::vector<float>& sample)
{
    m_pImpl->enqueueSample(sample);
}
 
//=============================================================================================================
 
void stream_outlet::push_chunk(const std::vector<std::vector<float>>& chunk)
{
    m_pImpl->enqueueChunk(chunk);
}
 
//=============================================================================================================
 
stream_info stream_outlet::info() const
{
    return m_pImpl->info();
}
 
//=============================================================================================================
 
bool stream_outlet::have_consumers() const
{
    return m_pImpl->haveConsumers();
}