From 1fba7ba314e972b2605acbdce08564357406cee0 Mon Sep 17 00:00:00 2001
From: f4exb <f4exb06@gmail.com>
Date: Sun, 30 Aug 2015 02:19:59 +0200
Subject: [PATCH] Deep redesign: fixed message dequeing so that non processed
 messages stay in queue

---
 sdrbase/dsp/dspengine.cpp     | 22 ++++++++++++++++------
 sdrbase/dsp/samplesink.cpp    |  9 ++++++++-
 sdrbase/dsp/samplesource.cpp  |  9 ++++++++-
 sdrbase/mainwindow.cpp        | 26 ++++++++++++++++++--------
 sdrbase/util/messagequeue.cpp |  2 ++
 5 files changed, 52 insertions(+), 16 deletions(-)

diff --git a/sdrbase/dsp/dspengine.cpp b/sdrbase/dsp/dspengine.cpp
index b256f3b74..584728167 100644
--- a/sdrbase/dsp/dspengine.cpp
+++ b/sdrbase/dsp/dspengine.cpp
@@ -608,13 +608,13 @@ void DSPEngine::handleSynchronousMessages()
 
 void DSPEngine::handleInputMessages()
 {
-	qDebug() << "DSPEngine::handleInputMessages";
-
 	Message* message;
+	int queueSize =  m_inputMessageQueue.size();
 
-	while ((message = m_inputMessageQueue.pop()) != 0)
+	for (int i = 0; i < queueSize; i++)
 	{
-		qDebug("DSPEngine::handleInputMessages: message: %s", message->getIdentifier());
+		message = m_inputMessageQueue.pop();
+		qDebug() << "DSPEngine::handleInputMessages: " << message->getIdentifier();
 
 		if (DSPConfigureCorrection::match(*message))
 		{
@@ -636,7 +636,11 @@ void DSPEngine::handleInputMessages()
 				m_imbalance = 65536;
 			}
 
-			delete message;
+			delete message; // delete
+		}
+		else
+		{
+			m_inputMessageQueue.push(message); // repush
 		}
 	}
 }
@@ -644,9 +648,11 @@ void DSPEngine::handleInputMessages()
 void DSPEngine::handleSourceMessages()
 {
 	Message *message;
+	int queueSize =  m_inputMessageQueue.size();
 
-	while ((message = m_sampleSource->getOutputMessageQueue()->pop()) != 0)
+	for (int i = 0; i < queueSize; i++)
 	{
+		message = m_inputMessageQueue.pop();
 		qDebug() << "DSPEngine::handleSourceMessages: " << message->getIdentifier();
 
 		if (DSPSignalNotification::match(*message))
@@ -683,6 +689,10 @@ void DSPEngine::handleSourceMessages()
 
 			delete message;
 		}
+		else
+		{
+			m_inputMessageQueue.push(message);
+		}
 	}
 }
 
diff --git a/sdrbase/dsp/samplesink.cpp b/sdrbase/dsp/samplesink.cpp
index 242c32e83..0aed61735 100644
--- a/sdrbase/dsp/samplesink.cpp
+++ b/sdrbase/dsp/samplesink.cpp
@@ -13,13 +13,20 @@ SampleSink::~SampleSink()
 void SampleSink::handleInputMessages()
 {
 	Message* message;
+	int queueSize = m_inputMessageQueue.size();
 
-	while ((message = m_inputMessageQueue.pop()) != 0)
+	for (int i = 0; i < queueSize; i++)
 	{
+		message = m_inputMessageQueue.pop();
+
 		if (handleMessage(*message))
 		{
 			delete message;
 		}
+		else
+		{
+			m_inputMessageQueue.push(message);
+		}
 	}
 }
 
diff --git a/sdrbase/dsp/samplesource.cpp b/sdrbase/dsp/samplesource.cpp
index be4958178..ea25c31c0 100644
--- a/sdrbase/dsp/samplesource.cpp
+++ b/sdrbase/dsp/samplesource.cpp
@@ -29,12 +29,19 @@ SampleSource::~SampleSource()
 void SampleSource::handleInputMessages()
 {
 	Message* message;
+	int queueSize = m_inputMessageQueue.size();
 
-	while ((message = m_inputMessageQueue.pop()) != 0)
+	for (int i = 0; i < queueSize; i++)
 	{
+		message = m_inputMessageQueue.pop();
+
 		if (handleMessage(*message))
 		{
 			delete message;
 		}
+		else
+		{
+			m_inputMessageQueue.push(message);
+		}
 	}
 }
diff --git a/sdrbase/mainwindow.cpp b/sdrbase/mainwindow.cpp
index a799172a3..b5cd7ee59 100644
--- a/sdrbase/mainwindow.cpp
+++ b/sdrbase/mainwindow.cpp
@@ -332,12 +332,13 @@ void MainWindow::applySettings()
 void MainWindow::handleDSPMessages()
 {
 	Message* message;
+	MessageQueue* messageQueue = m_dspEngine->getOutputMessageQueue();
+	int queueSize =  messageQueue->size();
 
-	while ((message = m_dspEngine->getOutputMessageQueue()->pop()) != 0)
+	for (int i = 0; i < queueSize; i++)
 	{
-		qDebug("Message: %s", message->getIdentifier());
-
-		std::cerr << "MainWindow::handleDSPMessages: " << message->getIdentifier() << std::endl;
+		message = messageQueue->pop();
+		qDebug() << "MainWindow::handleDSPMessages: " << message->getIdentifier();
 
 		if (DSPSignalNotification::match(*message))
 		{
@@ -352,22 +353,31 @@ void MainWindow::handleDSPMessages()
 
 			delete message;
 		}
+		else
+		{
+			messageQueue->push(message);
+		}
 	}
 }
 
 void MainWindow::handleMessages()
 {
 	Message* message;
+	int queueSize =  m_inputMessageQueue.size();
 
-	while ((message = m_inputMessageQueue.pop()) != 0)
+	for (int i = 0; i < queueSize; i++)
 	{
-		qDebug("Message: %s", message->getIdentifier());
-		std::cerr << "MainWindow::handleMessages: " << message->getIdentifier() << std::endl;
+		message = m_inputMessageQueue.pop();
+		qDebug() << "MainWindow::handleMessages: " << message->getIdentifier();
 
-		if (!m_pluginManager->handleMessage(*message))
+		if (m_pluginManager->handleMessage(*message))
 		{
 			delete message;
 		}
+		else
+		{
+			m_inputMessageQueue.push(message);
+		}
 	}
 }
 
diff --git a/sdrbase/util/messagequeue.cpp b/sdrbase/util/messagequeue.cpp
index 01f3b1817..846c640f7 100644
--- a/sdrbase/util/messagequeue.cpp
+++ b/sdrbase/util/messagequeue.cpp
@@ -15,6 +15,7 @@
 // along with this program. If not, see <http://www.gnu.org/licenses/>.          //
 ///////////////////////////////////////////////////////////////////////////////////
 
+#include <QDebug>
 #include "util/messagequeue.h"
 #include "util/message.h"
 
@@ -31,6 +32,7 @@ MessageQueue::~MessageQueue()
 
 	while ((message = pop()) != 0)
 	{
+		qDebug() << "MessageQueue::~MessageQueue: message: " << message->getIdentifier() << " was still in queue";
 		delete message;
 	}
 }