From 67847d3fceb664d593e1765c752edea122b8c365 Mon Sep 17 00:00:00 2001 From: IGNNE <30634883+IGNNE@users.noreply.github.com> Date: Thu, 29 Jul 2021 11:53:37 +0200 Subject: [PATCH 1/4] Fix compilation on newer Qt by adding an include --- app/generator/uianalogshape.cpp | 1 + 1 file changed, 1 insertion(+) diff --git a/app/generator/uianalogshape.cpp b/app/generator/uianalogshape.cpp index 64c1d5b..d927182 100644 --- a/app/generator/uianalogshape.cpp +++ b/app/generator/uianalogshape.cpp @@ -17,6 +17,7 @@ #include #include +#include /*! \class UiAnalogShape From 6b4a07621218eff0a5e38d45099241a5c28275c9 Mon Sep 17 00:00:00 2001 From: IGNNE <30634883+IGNNE@users.noreply.github.com> Date: Thu, 29 Jul 2021 12:11:48 +0200 Subject: [PATCH 2/4] Added network streaming feature --- app/LabTool.pro | 4 +- app/capture/captureapp.cpp | 62 ++++++++ app/capture/captureapp.h | 6 + app/capture/uicapturestreamer.cpp | 256 ++++++++++++++++++++++++++++++ app/capture/uicapturestreamer.h | 88 ++++++++++ 5 files changed, 415 insertions(+), 1 deletion(-) create mode 100644 app/capture/uicapturestreamer.cpp create mode 100644 app/capture/uicapturestreamer.h diff --git a/app/LabTool.pro b/app/LabTool.pro index 9d3247d..564e7ae 100644 --- a/app/LabTool.pro +++ b/app/LabTool.pro @@ -1,4 +1,5 @@ SOURCES += \ + capture/uicapturestreamer.cpp \ main.cpp \ generator/i2cgenerator.cpp \ device/devicemanager.cpp \ @@ -75,6 +76,7 @@ SOURCES += \ device/reconfigurelistener.cpp HEADERS += \ + capture/uicapturestreamer.h \ generator/i2cgenerator.h \ libusbx/include/libusbx-1.0/libusb.h \ device/devicemanager.h \ @@ -164,7 +166,7 @@ else:unix:!symbian: LIBS += -L$$PWD/libusbx/Linux/ -lusb-1.0 -ludev INCLUDEPATH += $$PWD/libusbx/MS32/dll DEPENDPATH += $$PWD/libusbx/MS32/dll -QT += widgets +QT += widgets network mac { ICON = resources/oscilloscope.icns diff --git a/app/capture/captureapp.cpp b/app/capture/captureapp.cpp index bc255ed..62aa048 100644 --- a/app/capture/captureapp.cpp +++ b/app/capture/captureapp.cpp @@ -55,6 +55,7 @@ CaptureApp::CaptureApp(QWidget* uiContext, QObject *parent) : mUiContext = uiContext; mCaptureActive = false; + mContinuous = false; // Deallocation: uiContext is set as parent mArea = new UiCaptureArea(mSignalManager, uiContext); @@ -75,6 +76,10 @@ CaptureApp::CaptureApp(QWidget* uiContext, QObject *parent) : } } + + mStreamingActive = false; + captureStreamer = new UiCaptureStreamer(DeviceManager::instance().activeDevice()->captureDevice(), mUiContext); + } /*! @@ -275,6 +280,8 @@ void CaptureApp::saveProject(QSettings &project) */ void CaptureApp::handleDeviceChanged(Device* activeDevice) { + delete captureStreamer; + captureStreamer = new UiCaptureStreamer(activeDevice->captureDevice(), mUiContext); setupRates(activeDevice->captureDevice()); mSignalManager->reloadSignalsFromDevice(); mArea->updateAnalogGroup(); @@ -454,6 +461,19 @@ void CaptureApp::createMenu() connect(action, SIGNAL(triggered()), this, SLOT(exportData())); mMenu->addAction(action); + + // + // Set Up Socket + // + + // Deallocation: "Qt Object trees" (See UiMainWindow) + mStreamAction = new QAction(tr("Stream Data to Socket"), this); + mStreamAction->setData("Stream Data to Socket"); + mStreamAction->setToolTip("Open a socket and send the currently captured data there"); + connect(mStreamAction, SIGNAL(triggered()), this, SLOT(streamData())); + mMenu->addAction(mStreamAction); + + } /*! @@ -747,6 +767,48 @@ void CaptureApp::exportData() } +/*! + Called when the user selects to stream data to socket, adapted from exportData +*/ +void CaptureApp::streamData() +{ + if(mStreamingActive) { + // currently streaming, so stop now + captureStreamer->stopStreaming(); + mStreamingActive = false; + mStreamAction->setText(tr("Stream Data to Socket")); + mStreamAction->setData("Stream Data to Socket"); + + } else { + // currently not streaming, so (try to) start now + + // checks before streaming + CaptureDevice* device = DeviceManager::instance().activeDevice() + ->captureDevice(); + if (device == NULL) { + return; + } + // check if there is data to stream + if(device->digitalSignals().empty() && device->digitalSignals().empty()) { + QMessageBox::warning(mUiContext, + "No signal found", + "Please add at least one signal!"); + return; + } + + + if(captureStreamer->exec() != QDialog::Accepted) { + // not accepted, abort + return; + } + + mStreamingActive = true; + mStreamAction->setText(tr("Stop Streaming")); + mStreamAction->setData("Stop Streaming"); + } + +} + /*! Called when the sample rate has changed. */ diff --git a/app/capture/captureapp.h b/app/capture/captureapp.h index 65aa1ff..e81dbf5 100644 --- a/app/capture/captureapp.h +++ b/app/capture/captureapp.h @@ -24,6 +24,7 @@ #include #include "uicapturearea.h" +#include "uicapturestreamer.h" #include "device/device.h" class CaptureApp : public QObject @@ -65,11 +66,15 @@ public slots: QAction* mTbStartAction; QAction* mTbContinuousAction; QAction* mTbStopAction; + QAction* mStreamAction; QComboBox* mRateBox; bool mCaptureActive; + bool mStreamingActive; + UiCaptureStreamer* captureStreamer; + void createToolBar(); void createMenu(); void changeCaptureActions(bool captureActive); @@ -87,6 +92,7 @@ private slots: void calibrationSettings(); void selectSignalsToAdd(); void exportData(); + void streamData(); void sampleRateChanged(int rateIndex); diff --git a/app/capture/uicapturestreamer.cpp b/app/capture/uicapturestreamer.cpp new file mode 100644 index 0000000..bfcdb5b --- /dev/null +++ b/app/capture/uicapturestreamer.cpp @@ -0,0 +1,256 @@ +#include "uicapturestreamer.h" + +#include +#include +#include +#include + + +/*! + \class UiCaptureStreamer + \brief This class is responsible for setting up and managing the data streaming. + + \ingroup Capture + +*/ + +/*! + Dialog Window +*/ +UiCaptureStreamer::UiCaptureStreamer(CaptureDevice* device, QWidget *parent) : + QDialog(parent) +{ + setWindowTitle(tr("Stream Data")); + setWindowFlags(windowFlags() & ~Qt::WindowContextHelpButtonHint); + + mCaptureDevice = device; + + // Deallocation: Ownership changed when calling setLayout. + mMainLayout = new QVBoxLayout(); + + // Deallocation: Re-parented when calling mMainLayout->addLayout. + QFormLayout* formLayout = new QFormLayout(); + + // Deallocation: "Qt Object trees" (See UiMainWindow) + mPortSpinBox = new QSpinBox(this); + mPortSpinBox->setMinimum(1); + mPortSpinBox->setMaximum(65535); + mPortSpinBox->setValue(18080); // random port + formLayout->addRow(tr("Port: "), mPortSpinBox); + + mMainLayout->addLayout(formLayout); + + // Deallocation: "Qt Object trees" (See UiMainWindow) + QPushButton* streamBtn = new QPushButton("Stream", this); + connect(streamBtn, SIGNAL(clicked()), this, SLOT(streamData())); + + QPushButton* cancelBtn = new QPushButton("Cancel", this); + connect(cancelBtn, SIGNAL(clicked()), this, SLOT(reject())); + + // Deallocation: Re-parented when calling mMainLayout->addLayout. + QHBoxLayout* hLayout = new QHBoxLayout(); + hLayout->addWidget(streamBtn); + hLayout->addWidget(cancelBtn); + hLayout->addStretch(); + mMainLayout->addLayout(hLayout); + + mMainLayout->addStretch(); + + setLayout(mMainLayout); + + // set up and connect StreamWorker + streamWorker.moveToThread(&workerThread); + streamWorker.setCaptureDevice(device); + connect(this, &UiCaptureStreamer::start, &streamWorker, &StreamWorker::start); + connect(this, &UiCaptureStreamer::stop, &streamWorker, &StreamWorker::stop); + connect(device, &CaptureDevice::captureFinished, &streamWorker, &StreamWorker::handleCaptureFinished); + +} + +/*! + Called when the user clicks the Stream button. +*/ +void UiCaptureStreamer::streamData() +{ + // check if the worker is in the right state + if(streamWorker.getState() != StreamWorker::STOPPED) { + reject(); + return; + } + + // start thread if necessary + if(!workerThread.isRunning()) { + workerThread.start(); + } + + emit start(mPortSpinBox->value()); + + // check if Worker is done initializing or timeout, a bit ugly + for (int i = 0; i < 50 && streamWorker.getState() == StreamWorker::STOPPED; i++) { + streamWorker.stateChangingMutex.tryLock(100); + } + if(streamWorker.getState() == StreamWorker::STOPPED) { + QMessageBox::warning(this, + "Stream Error", + "Timeout when setting up server, aborting"); + // for good measure, try to close the server again + emit stop(); + reject(); + return; + } + if(streamWorker.getState() == StreamWorker::ERROR) { + QMessageBox::warning(this, + "Stream Error", + "Failed to set up server, check port!"); + // this sets the state back to stopped + emit stop(); + reject(); + return; + } + + accept(); +} + +/*! + Waits until the streaming is stopped +*/ +void UiCaptureStreamer::stopStreaming() { + emit stop(); + while(streamWorker.getState() != StreamWorker::STOPPED) { + streamWorker.stateChangingMutex.tryLock(1000); + } +} + +/////////////////////////////////////////////////////////////////////////////// +// StreamWorker Implementation below +/////////////////////////////////////////////////////////////////////////////// + +/*! + * Start streaming (start listening) + */ +void StreamWorker::start(int port) { + if(port < 1 || port > 65535) { + // bad port value + state = StreamingState::ERROR; + return; + } + + if(state != StreamingState::STOPPED) { + // we are not in the right state, stop and indicate an error + stop(); + state = StreamingState::ERROR; + return; + } + + if (server == nullptr) { + server = new QTcpServer(); + } + + if(!server->listen(QHostAddress::Any, port)) { + // error with the port + state = StreamingState::ERROR; + server->close(); + return; + } + + connect(server, &QTcpServer::newConnection, this, &StreamWorker::handleNewConnection); + qInfo() << "StreamWorker: Starting"; + state = StreamingState::RUNNING; +} + +/*! + * Stop streaming + */ +void StreamWorker::stop() { + qInfo() << "StreamWorker: Stopping"; + if(state == StreamingState::RUNNING) { + server->close(); + sockets.clear(); + } + state = StreamingState::STOPPED; +} + +/*! + Handle a new connection by adding it to the list and remove it when disconnected + \todo check if this works +*/ +void StreamWorker::handleNewConnection() { + QTcpSocket* socket = server->nextPendingConnection(); + if(socket == nullptr) { + // I have no idea why this would happen, but it happens + return; + } + sockets.append(socket); + qInfo() << "StreamWorker: Got new connection"; + connect(socket, &QTcpSocket::disconnected, [this, socket](){ + this->sockets.removeOne(socket); + // delete socket; + // maybe there is a memory leak here, but with the delete, it will crash + } ); +} + +/*! + Send the captured data to all active sockets + \todo check if this works +*/ +void StreamWorker::handleCaptureFinished(bool successful, QString msg) { + qDebug() << "StreamWorker: Got new data, sending to " << sockets.length() << " clients"; + if(state == StreamingState::RUNNING && successful) { + QJsonObject json; + this->writeToJson(json); + // write compact json, so one message -> one line + // this simplifies reading in clients with tcp streaming + // so the delimiter between messages will be '\n' + QByteArray jsonBytes = QJsonDocument(json).toJson(QJsonDocument::Compact); + foreach(QTcpSocket* socket, sockets) { + socket->write(jsonBytes); + socket->write("\n"); + } + } +} + +/*! + Writes the current state of the CaptureDevice to JSON + */ +void StreamWorker::writeToJson(QJsonObject &json) { + QJsonArray jsonSignals; + + QList digitalSignals = device->digitalSignals(); + QList analogSignals = device->analogSignals(); + + json.insert("sampleRate", device->usedSampleRate()); + + foreach(DigitalSignal* s, digitalSignals) { + QVector* data = device->digitalData(s->id()); + if (data == NULL) continue; + + QJsonObject jsonSignal; + jsonSignal.insert("id", s->id()); + jsonSignal.insert("name", s->name()); + jsonSignal.insert("type", "digital"); + QJsonArray jsonData; + foreach(int dataPoint, *data) { + jsonData.append(dataPoint); + } + jsonSignal.insert("data", jsonData); + jsonSignals.append(jsonSignal); + } + foreach(AnalogSignal* s, analogSignals) { + QVector* data = device->analogData(s->id()); + if (data == NULL) continue; + + QJsonObject jsonSignal; + jsonSignal.insert("id", s->id()); + jsonSignal.insert("name", s->name()); + jsonSignal.insert("type", "analog"); + QJsonArray jsonData; + foreach(double dataPoint, *data) { + jsonData.append(dataPoint); + } + jsonSignal.insert("data", jsonData); + jsonSignals.append(jsonSignal); + } + + json.insert("signals", jsonSignals); + +} diff --git a/app/capture/uicapturestreamer.h b/app/capture/uicapturestreamer.h new file mode 100644 index 0000000..3f70d8d --- /dev/null +++ b/app/capture/uicapturestreamer.h @@ -0,0 +1,88 @@ +#ifndef UICAPTURESTREAMER_H +#define UICAPTURESTREAMER_H + +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +#include "device/capturedevice.h" + +/*! + \class StreamWorker + \brief Worker class/thread managed by \a UiCaptureStreamer + + \ingroup Capture +*/ +class StreamWorker : public QObject { + Q_OBJECT + +public: + enum StreamingState { + STOPPED, // not running, no current error + RUNNING, // successfully running + ERROR // not running because of an error + }; + QMutex stateChangingMutex; // indicate that a (possible) state change is happening + StreamingState getState() { return state; }; + void setCaptureDevice(CaptureDevice* device) { this->device = device; }; + +private: + QTcpServer *server = nullptr; + StreamingState state = StreamingState::STOPPED; + CaptureDevice* device; + + QList sockets; + + void writeToJson(QJsonObject &json); + +private slots: + void handleNewConnection(); +public slots: + void start(int port); + void stop(); + void handleCaptureFinished(bool successful, QString msg); + +}; + +/*! + \class UiCaptureStreamer + \brief This class is responsible for setting up and managing the data streaming. + Largely an adapted form of \a UiCaptureExporter + + \ingroup Capture +*/ +class UiCaptureStreamer : public QDialog +{ + Q_OBJECT +public: + explicit UiCaptureStreamer(CaptureDevice* device, QWidget *parent = 0); + void stopStreaming(); + +private: + + QThread workerThread; + StreamWorker streamWorker; + + CaptureDevice* mCaptureDevice; + QVBoxLayout* mMainLayout; + + QSpinBox* mPortSpinBox; + +private slots: + void streamData(); + +signals: + void start(int port); + void stop(); + +}; + +#endif // UICAPTURESTREAMER_H From 8f4f26d78cfb2edc936a8f12790122b3297649a3 Mon Sep 17 00:00:00 2001 From: IGNNE <30634883+IGNNE@users.noreply.github.com> Date: Sat, 31 Jul 2021 01:50:38 +0200 Subject: [PATCH 3/4] Rewrite and cleanup of the streaming code --- app/LabTool.pro | 2 + app/capture/captureapp.cpp | 8 +- app/capture/streamworker.cpp | 164 ++++++++++++++++++++++ app/capture/streamworker.h | 66 +++++++++ app/capture/uicapturestreamer.cpp | 223 ++++++------------------------ app/capture/uicapturestreamer.h | 61 ++------ 6 files changed, 296 insertions(+), 228 deletions(-) create mode 100644 app/capture/streamworker.cpp create mode 100644 app/capture/streamworker.h diff --git a/app/LabTool.pro b/app/LabTool.pro index 564e7ae..a202ff8 100644 --- a/app/LabTool.pro +++ b/app/LabTool.pro @@ -1,4 +1,5 @@ SOURCES += \ + capture/streamworker.cpp \ capture/uicapturestreamer.cpp \ main.cpp \ generator/i2cgenerator.cpp \ @@ -76,6 +77,7 @@ SOURCES += \ device/reconfigurelistener.cpp HEADERS += \ + capture/streamworker.h \ capture/uicapturestreamer.h \ generator/i2cgenerator.h \ libusbx/include/libusbx-1.0/libusb.h \ diff --git a/app/capture/captureapp.cpp b/app/capture/captureapp.cpp index 62aa048..1a73836 100644 --- a/app/capture/captureapp.cpp +++ b/app/capture/captureapp.cpp @@ -280,8 +280,14 @@ void CaptureApp::saveProject(QSettings &project) */ void CaptureApp::handleDeviceChanged(Device* activeDevice) { + // recreate captureStreamer (even if not running) delete captureStreamer; captureStreamer = new UiCaptureStreamer(activeDevice->captureDevice(), mUiContext); + // update UI to follow up + // this will make it end up in the "stopped" UI state + mStreamingActive = true; + streamData(); + setupRates(activeDevice->captureDevice()); mSignalManager->reloadSignalsFromDevice(); mArea->updateAnalogGroup(); @@ -774,7 +780,7 @@ void CaptureApp::streamData() { if(mStreamingActive) { // currently streaming, so stop now - captureStreamer->stopStreaming(); + emit captureStreamer->stopWorker(); mStreamingActive = false; mStreamAction->setText(tr("Stream Data to Socket")); mStreamAction->setData("Stream Data to Socket"); diff --git a/app/capture/streamworker.cpp b/app/capture/streamworker.cpp new file mode 100644 index 0000000..780d400 --- /dev/null +++ b/app/capture/streamworker.cpp @@ -0,0 +1,164 @@ +#include "streamworker.h" + +StreamWorker::StreamWorker(CaptureDevice* device) : QObject(nullptr) { + this->device = device; +// server = new QTcpServer(); +// connect(server, &QTcpServer::newConnection, this, &StreamWorker::handleNewConnection); +} + +StreamWorker::~StreamWorker() +{ + stop(); + delete server; +} + +/*! +Start streaming (start listening) +*/ +void StreamWorker::start(int port) { + if(server == nullptr) { + // the server does not like to be dragged across threads, so create it only when we need it (and already are in the right thread) + server = new QTcpServer(); + connect(server, &QTcpServer::newConnection, this, &StreamWorker::handleNewConnection); + } + if(port < 1 || port > 65535) { + // bad port value + setState(StreamingState::ERROR); + return; + } + + if(getState() != StreamingState::STOPPED) { + // we are not in the right state, stop and indicate an error + stop(); + setState(StreamingState::ERROR); + return; + } + + if(!server->listen(QHostAddress::Any, port)) { + // error with the port + setState(StreamingState::ERROR); + server->close(); + return; + } + + qInfo() << "StreamWorker: Starting"; + setState(StreamingState::RUNNING); +} + +/*! + Stop streaming + */ +void StreamWorker::stop() { + qInfo() << "StreamWorker: Stopping"; + if(getState() == StreamingState::RUNNING) { + server->close(); + foreach (auto socket, sockets) { + socket->deleteLater(); + } + sockets.clear(); + } + setState(StreamingState::STOPPED); +} + +/*! +Sets the state, synchronizes with a mutex, and emits the appropriate signal + */ +void StreamWorker::setState(StreamingState newState) { + QMutexLocker stateChangingMutexLocker {&stateChangingMutex}; + state = newState; + switch (state) { + case STOPPED: + emit stopped(); + break; + case RUNNING: + emit running(); + break; + case ERROR: + emit error(); + break; + }; +} + +/*! + Handle a new connection by adding it to the list and remove it when disconnected +*/ +void StreamWorker::handleNewConnection() { + QTcpSocket* socket = server->nextPendingConnection(); + if(socket == nullptr) { + // I have no idea why this would happen, but it happens + return; + } + sockets.append(socket); + qInfo() << "StreamWorker: Got new connection: " << (long)socket; + connect(socket, &QTcpSocket::disconnected, [this, socket](){ + this->sockets.removeOne(socket); + qDebug() << "StreamWorker: Removed a connection: " << (long)socket; + socket->deleteLater(); + } ); +} + +/*! + Send the captured data to all active sockets +*/ +void StreamWorker::handleCaptureFinished(bool successful, QString msg) { + // no need to further synchronize this, since all of this class is run via the event loop of its thread + if(getState() == StreamingState::RUNNING && successful) { + qDebug() << "StreamWorker: Got new data, sending to " << sockets.length() << " clients"; + QJsonObject json; + this->writeToJson(json); + // write compact json, so one message -> one line + // this simplifies reading in clients with tcp streaming + // so the delimiter between messages will be '\n' + QByteArray jsonBytes = QJsonDocument(json).toJson(QJsonDocument::Compact); + foreach(QTcpSocket* socket, sockets) { + socket->write(jsonBytes); + socket->write("\n"); + } + } +} + +/*! + Writes the current state of the CaptureDevice to JSON +*/ +void StreamWorker::writeToJson(QJsonObject &json) { + QJsonArray jsonSignals; + + QList digitalSignals = device->digitalSignals(); + QList analogSignals = device->analogSignals(); + + json.insert("sampleRate", device->usedSampleRate()); + + foreach(DigitalSignal* s, digitalSignals) { + QVector* data = device->digitalData(s->id()); + if (data == NULL) continue; + + QJsonObject jsonSignal; + jsonSignal.insert("id", s->id()); + jsonSignal.insert("name", s->name()); + jsonSignal.insert("type", "digital"); + QJsonArray jsonData; + foreach(int dataPoint, *data) { + jsonData.append(dataPoint); + } + jsonSignal.insert("data", jsonData); + jsonSignals.append(jsonSignal); + } + foreach(AnalogSignal* s, analogSignals) { + QVector* data = device->analogData(s->id()); + if (data == NULL) continue; + + QJsonObject jsonSignal; + jsonSignal.insert("id", s->id()); + jsonSignal.insert("name", s->name()); + jsonSignal.insert("type", "analog"); + QJsonArray jsonData; + foreach(double dataPoint, *data) { + jsonData.append(dataPoint); + } + jsonSignal.insert("data", jsonData); + jsonSignals.append(jsonSignal); + } + + json.insert("signals", jsonSignals); + +} diff --git a/app/capture/streamworker.h b/app/capture/streamworker.h new file mode 100644 index 0000000..90214b6 --- /dev/null +++ b/app/capture/streamworker.h @@ -0,0 +1,66 @@ +#ifndef STREAMWORKER_H +#define STREAMWORKER_H + +#include +#include +#include +#include +#include +#include +#include + +#include "device/capturedevice.h" + +/*! + \class StreamWorker + \brief Worker class/thread managed by \a UiCaptureStreamer + + \ingroup Capture +*/ +class StreamWorker : public QObject { + Q_OBJECT + +public: + StreamWorker(CaptureDevice* device); + ~StreamWorker(); + + enum StreamingState { + STOPPED, // not running, no current error + RUNNING, // successfully running + ERROR // not running because of an error + }; + + /// makes sure that this synchronized by a mutex + /// probably a bit paranoid, but doesn't hurt + StreamingState getState() { QMutexLocker stateChangingMutexLocker {&stateChangingMutex}; return state; }; + +private: + QMutex stateChangingMutex; // indicate that a (possible) state change is happening + QTcpServer *server = nullptr; + StreamingState state = StreamingState::STOPPED; + CaptureDevice* device = nullptr; + + QList sockets; + + void writeToJson(QJsonObject &json); + + /// makes sure that this synchronized by a mutex and also emits the signal every time + void setState(StreamingState newState); + +private slots: + void handleNewConnection(); + +public slots: + void start(int port); + void stop(); + void handleCaptureFinished(bool successful, QString msg); + +signals: + void running(); + void stopped(); + void error(); + void deleted(); + +}; + +#endif // STREAMWORKER_H diff --git a/app/capture/uicapturestreamer.cpp b/app/capture/uicapturestreamer.cpp index bfcdb5b..96c71e4 100644 --- a/app/capture/uicapturestreamer.cpp +++ b/app/capture/uicapturestreamer.cpp @@ -6,21 +6,13 @@ #include -/*! - \class UiCaptureStreamer - \brief This class is responsible for setting up and managing the data streaming. - - \ingroup Capture - -*/ - /*! Dialog Window */ UiCaptureStreamer::UiCaptureStreamer(CaptureDevice* device, QWidget *parent) : QDialog(parent) { - setWindowTitle(tr("Stream Data")); + setWindowTitle(tr("Stream")); setWindowFlags(windowFlags() & ~Qt::WindowContextHelpButtonHint); mCaptureDevice = device; @@ -28,9 +20,11 @@ UiCaptureStreamer::UiCaptureStreamer(CaptureDevice* device, QWidget *parent) : // Deallocation: Ownership changed when calling setLayout. mMainLayout = new QVBoxLayout(); - // Deallocation: Re-parented when calling mMainLayout->addLayout. QFormLayout* formLayout = new QFormLayout(); + mMainLayout->addWidget(new QLabel(tr("Stream data over TCP as JSON"), this)); + mMainLayout->addWidget(new QLabel(tr("Every message is in a new line"), this)); + // Deallocation: "Qt Object trees" (See UiMainWindow) mPortSpinBox = new QSpinBox(this); mPortSpinBox->setMinimum(1); @@ -42,7 +36,7 @@ UiCaptureStreamer::UiCaptureStreamer(CaptureDevice* device, QWidget *parent) : // Deallocation: "Qt Object trees" (See UiMainWindow) QPushButton* streamBtn = new QPushButton("Stream", this); - connect(streamBtn, SIGNAL(clicked()), this, SLOT(streamData())); + connect(streamBtn, SIGNAL(clicked()), this, SLOT(handleStreamBtnPressed())); QPushButton* cancelBtn = new QPushButton("Cancel", this); connect(cancelBtn, SIGNAL(clicked()), this, SLOT(reject())); @@ -59,198 +53,69 @@ UiCaptureStreamer::UiCaptureStreamer(CaptureDevice* device, QWidget *parent) : setLayout(mMainLayout); // set up and connect StreamWorker - streamWorker.moveToThread(&workerThread); - streamWorker.setCaptureDevice(device); - connect(this, &UiCaptureStreamer::start, &streamWorker, &StreamWorker::start); - connect(this, &UiCaptureStreamer::stop, &streamWorker, &StreamWorker::stop); - connect(device, &CaptureDevice::captureFinished, &streamWorker, &StreamWorker::handleCaptureFinished); + streamWorker = new StreamWorker(device); + workerThread = new QThread(); + streamWorker->moveToThread(workerThread); + connect(this, &UiCaptureStreamer::startWorker, streamWorker, &StreamWorker::start); + connect(this, &UiCaptureStreamer::stopWorker, streamWorker, &StreamWorker::stop); + connect(streamWorker, &StreamWorker::running, this, &UiCaptureStreamer::handleStreamRunning); + connect(streamWorker, &StreamWorker::error, this, &UiCaptureStreamer::handleStreamError); + connect(device, &CaptureDevice::captureFinished, streamWorker, &StreamWorker::handleCaptureFinished); + // make sure to properly set up deletion across threads + connect(this, &UiCaptureStreamer::destroyWorker, streamWorker, &StreamWorker::deleteLater); + connect(streamWorker, &StreamWorker::deleted, workerThread, &QThread::deleteLater); } /*! - Called when the user clicks the Stream button. +Destructor that takes care of stopping the thread etc. */ -void UiCaptureStreamer::streamData() +UiCaptureStreamer::~UiCaptureStreamer() { - // check if the worker is in the right state - if(streamWorker.getState() != StreamWorker::STOPPED) { - reject(); - return; - } - - // start thread if necessary - if(!workerThread.isRunning()) { - workerThread.start(); - } - - emit start(mPortSpinBox->value()); + // this stops the worker and then frees its resources + // then the worker will detele its thread via signals + emit destroyWorker(); - // check if Worker is done initializing or timeout, a bit ugly - for (int i = 0; i < 50 && streamWorker.getState() == StreamWorker::STOPPED; i++) { - streamWorker.stateChangingMutex.tryLock(100); - } - if(streamWorker.getState() == StreamWorker::STOPPED) { - QMessageBox::warning(this, - "Stream Error", - "Timeout when setting up server, aborting"); - // for good measure, try to close the server again - emit stop(); - reject(); - return; - } - if(streamWorker.getState() == StreamWorker::ERROR) { - QMessageBox::warning(this, - "Stream Error", - "Failed to set up server, check port!"); - // this sets the state back to stopped - emit stop(); - reject(); - return; - } - - accept(); } /*! - Waits until the streaming is stopped +Called when the user clicks the Stream button. */ -void UiCaptureStreamer::stopStreaming() { - emit stop(); - while(streamWorker.getState() != StreamWorker::STOPPED) { - streamWorker.stateChangingMutex.tryLock(1000); - } -} - -/////////////////////////////////////////////////////////////////////////////// -// StreamWorker Implementation below -/////////////////////////////////////////////////////////////////////////////// - -/*! - * Start streaming (start listening) - */ -void StreamWorker::start(int port) { - if(port < 1 || port > 65535) { - // bad port value - state = StreamingState::ERROR; - return; - } - - if(state != StreamingState::STOPPED) { - // we are not in the right state, stop and indicate an error - stop(); - state = StreamingState::ERROR; +void UiCaptureStreamer::handleStreamBtnPressed() +{ + // check if the worker is in the right state + if(streamWorker->getState() != StreamWorker::STOPPED) { + reject(); return; } - if (server == nullptr) { - server = new QTcpServer(); - } - - if(!server->listen(QHostAddress::Any, port)) { - // error with the port - state = StreamingState::ERROR; - server->close(); - return; + // start thread if necessary + if(!workerThread->isRunning()) { + workerThread->start(); } - connect(server, &QTcpServer::newConnection, this, &StreamWorker::handleNewConnection); - qInfo() << "StreamWorker: Starting"; - state = StreamingState::RUNNING; -} + emit startWorker(mPortSpinBox->value()); -/*! - * Stop streaming - */ -void StreamWorker::stop() { - qInfo() << "StreamWorker: Stopping"; - if(state == StreamingState::RUNNING) { - server->close(); - sockets.clear(); - } - state = StreamingState::STOPPED; + // the answer is handled by handleStreamWorkerRunning/Error } /*! - Handle a new connection by adding it to the list and remove it when disconnected - \todo check if this works +This will get called after the start request and a successful start */ -void StreamWorker::handleNewConnection() { - QTcpSocket* socket = server->nextPendingConnection(); - if(socket == nullptr) { - // I have no idea why this would happen, but it happens - return; - } - sockets.append(socket); - qInfo() << "StreamWorker: Got new connection"; - connect(socket, &QTcpSocket::disconnected, [this, socket](){ - this->sockets.removeOne(socket); - // delete socket; - // maybe there is a memory leak here, but with the delete, it will crash - } ); +void UiCaptureStreamer::handleStreamRunning() { + // just accept the dialog, everything went well + accept(); } /*! - Send the captured data to all active sockets - \todo check if this works +This will get called after the start request and an error */ -void StreamWorker::handleCaptureFinished(bool successful, QString msg) { - qDebug() << "StreamWorker: Got new data, sending to " << sockets.length() << " clients"; - if(state == StreamingState::RUNNING && successful) { - QJsonObject json; - this->writeToJson(json); - // write compact json, so one message -> one line - // this simplifies reading in clients with tcp streaming - // so the delimiter between messages will be '\n' - QByteArray jsonBytes = QJsonDocument(json).toJson(QJsonDocument::Compact); - foreach(QTcpSocket* socket, sockets) { - socket->write(jsonBytes); - socket->write("\n"); - } - } +void UiCaptureStreamer::handleStreamError() { + QMessageBox::warning(this, + "Stream Error", + "Failed to set up server, check port!"); + // this sets the state back to stopped + emit stopWorker(); + reject(); } -/*! - Writes the current state of the CaptureDevice to JSON - */ -void StreamWorker::writeToJson(QJsonObject &json) { - QJsonArray jsonSignals; - - QList digitalSignals = device->digitalSignals(); - QList analogSignals = device->analogSignals(); - - json.insert("sampleRate", device->usedSampleRate()); - - foreach(DigitalSignal* s, digitalSignals) { - QVector* data = device->digitalData(s->id()); - if (data == NULL) continue; - - QJsonObject jsonSignal; - jsonSignal.insert("id", s->id()); - jsonSignal.insert("name", s->name()); - jsonSignal.insert("type", "digital"); - QJsonArray jsonData; - foreach(int dataPoint, *data) { - jsonData.append(dataPoint); - } - jsonSignal.insert("data", jsonData); - jsonSignals.append(jsonSignal); - } - foreach(AnalogSignal* s, analogSignals) { - QVector* data = device->analogData(s->id()); - if (data == NULL) continue; - - QJsonObject jsonSignal; - jsonSignal.insert("id", s->id()); - jsonSignal.insert("name", s->name()); - jsonSignal.insert("type", "analog"); - QJsonArray jsonData; - foreach(double dataPoint, *data) { - jsonData.append(dataPoint); - } - jsonSignal.insert("data", jsonData); - jsonSignals.append(jsonSignal); - } - - json.insert("signals", jsonSignals); - -} diff --git a/app/capture/uicapturestreamer.h b/app/capture/uicapturestreamer.h index 3f70d8d..4feabb3 100644 --- a/app/capture/uicapturestreamer.h +++ b/app/capture/uicapturestreamer.h @@ -5,52 +5,12 @@ #include #include #include -#include -#include -#include -#include -#include -#include -#include +#include -#include "device/capturedevice.h" - -/*! - \class StreamWorker - \brief Worker class/thread managed by \a UiCaptureStreamer - - \ingroup Capture -*/ -class StreamWorker : public QObject { - Q_OBJECT - -public: - enum StreamingState { - STOPPED, // not running, no current error - RUNNING, // successfully running - ERROR // not running because of an error - }; - QMutex stateChangingMutex; // indicate that a (possible) state change is happening - StreamingState getState() { return state; }; - void setCaptureDevice(CaptureDevice* device) { this->device = device; }; - -private: - QTcpServer *server = nullptr; - StreamingState state = StreamingState::STOPPED; - CaptureDevice* device; - - QList sockets; +#include "streamworker.h" - void writeToJson(QJsonObject &json); - -private slots: - void handleNewConnection(); -public slots: - void start(int port); - void stop(); - void handleCaptureFinished(bool successful, QString msg); +#include "device/capturedevice.h" -}; /*! \class UiCaptureStreamer @@ -64,12 +24,13 @@ class UiCaptureStreamer : public QDialog Q_OBJECT public: explicit UiCaptureStreamer(CaptureDevice* device, QWidget *parent = 0); + ~UiCaptureStreamer(); void stopStreaming(); private: - QThread workerThread; - StreamWorker streamWorker; + QThread* workerThread; + StreamWorker* streamWorker; CaptureDevice* mCaptureDevice; QVBoxLayout* mMainLayout; @@ -77,11 +38,15 @@ class UiCaptureStreamer : public QDialog QSpinBox* mPortSpinBox; private slots: - void streamData(); + void handleStreamBtnPressed(); + void handleStreamRunning(); + void handleStreamError(); signals: - void start(int port); - void stop(); + void startWorker(int port); + void stopWorker(); + void destroyWorker(); + void destroyThread(); }; From 809a713f5aeb625377ef77e6345d32ded6f3bc07 Mon Sep 17 00:00:00 2001 From: IGNNE <30634883+IGNNE@users.noreply.github.com> Date: Sat, 31 Jul 2021 02:10:28 +0200 Subject: [PATCH 4/4] Minor bugfix, some clarification --- app/capture/captureapp.cpp | 5 ++--- 1 file changed, 2 insertions(+), 3 deletions(-) diff --git a/app/capture/captureapp.cpp b/app/capture/captureapp.cpp index 1a73836..17db23c 100644 --- a/app/capture/captureapp.cpp +++ b/app/capture/captureapp.cpp @@ -469,10 +469,9 @@ void CaptureApp::createMenu() // - // Set Up Socket + // Set Up Streaming via Network // - // Deallocation: "Qt Object trees" (See UiMainWindow) mStreamAction = new QAction(tr("Stream Data to Socket"), this); mStreamAction->setData("Stream Data to Socket"); mStreamAction->setToolTip("Open a socket and send the currently captured data there"); @@ -795,7 +794,7 @@ void CaptureApp::streamData() return; } // check if there is data to stream - if(device->digitalSignals().empty() && device->digitalSignals().empty()) { + if(device->digitalSignals().empty() && device->analogSignals().empty()) { QMessageBox::warning(mUiContext, "No signal found", "Please add at least one signal!");