More terminate()-and-flush calls missed before

This commit is contained in:
vsonnier 2017-08-29 18:48:59 +02:00
parent dc9c67bec1
commit 3842cf087f
8 changed files with 37 additions and 2 deletions

View File

@ -25,8 +25,8 @@ public:
void onBindOutput(std::string name, ThreadQueueBasePtr threadQueue); void onBindOutput(std::string name, ThreadQueueBasePtr threadQueue);
void run(); virtual void run();
void terminate(); virtual void terminate();
void setMuted(bool state); void setMuted(bool state);
bool isMuted(); bool isMuted();

View File

@ -84,3 +84,9 @@ void FFTVisualDataThread::run() {
// std::cout << "FFT visual data thread done." << std::endl; // std::cout << "FFT visual data thread done." << std::endl;
} }
void FFTVisualDataThread::terminate() {
IOThread::terminate();
fftDistrib.flushQueues();
wproc.flushQueues();
}

View File

@ -18,6 +18,8 @@ public:
virtual void run(); virtual void run();
virtual void terminate();
protected: protected:
FFTDataDistributor fftDistrib; FFTDataDistributor fftDistrib;
DemodulatorThreadInputQueuePtr fftQueue = std::make_shared<DemodulatorThreadInputQueue>(); DemodulatorThreadInputQueuePtr fftQueue = std::make_shared<DemodulatorThreadInputQueue>();

View File

@ -28,3 +28,7 @@ void SpectrumVisualDataThread::run() {
// std::cout << "Spectrum visual data thread done." << std::endl; // std::cout << "Spectrum visual data thread done." << std::endl;
} }
void SpectrumVisualDataThread::terminate() {
IOThread::terminate();
sproc.flushQueues();
}

View File

@ -14,6 +14,8 @@ public:
virtual void run(); virtual void run();
virtual void terminate();
protected: protected:
SpectrumVisualProcessor sproc; SpectrumVisualProcessor sproc;
}; };

View File

@ -82,6 +82,17 @@ public:
outputs.erase(i); outputs.erase(i);
} }
} }
//Flush all queues, either input or outputs clearing their accumulated messages.
//this is purposefully non-blocking call.
void flushQueues() {
//DO NOT take the busy_update, we want a never blocking op how imperfect it could be.
input->flush();
for (auto single_output : outputs) {
single_output->flush();
}
}
//Call process() repeateadly until all available 'input' data is consumed. //Call process() repeateadly until all available 'input' data is consumed.
void run() { void run() {

View File

@ -564,6 +564,15 @@ void SDRThread::run() {
std::cout << "SDR thread done." << std::endl; std::cout << "SDR thread done." << std::endl;
} }
void SDRThread::terminate() {
IOThread::terminate();
SDRThreadIQDataQueuePtr iqDataOutQueue = std::static_pointer_cast<SDRThreadIQDataQueue>(getOutputQueue("IQDataOutput"));
if (iqDataOutQueue != nullptr) {
iqDataOutQueue->flush();
}
}
SDRDeviceInfo *SDRThread::getDevice() { SDRDeviceInfo *SDRThread::getDevice() {
return deviceInfo.load(); return deviceInfo.load();

View File

@ -61,6 +61,7 @@ public:
enum SDRThreadState { SDR_THREAD_MESSAGE, SDR_THREAD_INITIALIZED, SDR_THREAD_FAILED}; enum SDRThreadState { SDR_THREAD_MESSAGE, SDR_THREAD_INITIALIZED, SDR_THREAD_FAILED};
virtual void run(); virtual void run();
virtual void terminate();
SDRDeviceInfo *getDevice(); SDRDeviceInfo *getDevice();
void setDevice(SDRDeviceInfo *dev); void setDevice(SDRDeviceInfo *dev);