fix TCPSrc

include/dsp/samplesink.h

@@ -1,12 +1,13 @@
 #ifndef INCLUDE_SAMPLESINK_H
 #define INCLUDE_SAMPLESINK_H
 
+#include <QObject>
 #include "dsptypes.h"
 #include "util/export.h"
 
 class Message;
 
-class SDRANGELOVE_API SampleSink {
+class SDRANGELOVE_API SampleSink : public QObject {
 public:
 	SampleSink();
 	virtual ~SampleSink();
@@ -18,26 +19,3 @@ public:
 };
 
 #endif // INCLUDE_SAMPLESINK_H
-
-#if 0
-#ifndef INCLUDE_SAMPLESINK_H
-#define INCLUDE_SAMPLESINK_H
-
-#include "dsptypes.h"
-
-class SampleSink {
-public:
-	SampleSink();
-
-	virtual size_t workUnitSize() = 0;
-
-	void feed(SampleVector::const_iterator begin, SampleVector::const_iterator end);
-	virtual size_t work(SampleVector::const_iterator begin, SampleVector::const_iterator end) = 0;
-
-private:
-	SampleVector m_sinkBuffer;
-	size_t m_sinkBufferFill;
-};
-
-#endif // INCLUDE_SAMPLESINK_H
-#endif

include/dsp/threadedsamplesink.h

@@ -10,7 +10,7 @@
 class QThread;
 class SampleSink;
 
-class SDRANGELOVE_API ThreadedSampleSink : public QObject, public SampleSink {
+class SDRANGELOVE_API ThreadedSampleSink : public SampleSink {
 	Q_OBJECT
 
 public:

include/gui/rollupwidget.h

@@ -19,6 +19,7 @@ protected:
 	void resizeEvent(QResizeEvent* size);
 	void mousePressEvent(QMouseEvent* event);
 
+	bool event(QEvent* event);
 	bool eventFilter(QObject* object, QEvent* event);
 };
 

plugins/demod/tcpsrc/tcpsrc.cpp

@@ -1,28 +1,37 @@
+#include <QTcpServer>
+#include <QTcpSocket>
+#include <QThread>
 #include "tcpsrc.h"
+#include "tcpsrcgui.h"
 #include "dsp/dspcommands.h"
 
-MessageRegistrator TCPSrc::MsgConfigureTCPSrc::ID("MsgConfigureTCPSrc");
+MessageRegistrator TCPSrc::MsgTCPSrcConfigure::ID("MsgTCPSrcConfigure");
+MessageRegistrator TCPSrc::MsgTCPSrcConnection::ID("MsgTCPSrcConnection");
 
-TCPSrc::TCPSrc(SampleSink* spectrum)
+TCPSrc::TCPSrc(MessageQueue* uiMessageQueue, TCPSrcGUI* tcpSrcGUI, SampleSink* spectrum)
 {
 	m_inputSampleRate = 100000;
-	m_sampleFormat = 0;
+	m_sampleFormat = FormatS8;
 	m_outputSampleRate = 50000;
 	m_rfBandwidth = 50000;
 	m_tcpPort = 9999;
 	m_nco.setFreq(0, m_inputSampleRate);
 	m_interpolator.create(16, m_inputSampleRate, m_rfBandwidth / 2.1);
 	m_sampleDistanceRemain = m_inputSampleRate / m_outputSampleRate;
+	m_uiMessageQueue = uiMessageQueue;
+	m_tcpSrcGUI = tcpSrcGUI;
 	m_spectrum = spectrum;
+	m_nextS8Id = 0;
+	m_nextS16leId = 0;
 }
 
 TCPSrc::~TCPSrc()
 {
 }
 
-void TCPSrc::configure(MessageQueue* messageQueue, int sampleFormat, Real outputSampleRate, Real rfBandwidth, int tcpPort)
+void TCPSrc::configure(MessageQueue* messageQueue, SampleFormat sampleFormat, Real outputSampleRate, Real rfBandwidth, int tcpPort)
 {
-	Message* cmd = MsgConfigureTCPSrc::create(sampleFormat, outputSampleRate, rfBandwidth, tcpPort);
+	Message* cmd = MsgTCPSrcConfigure::create(sampleFormat, outputSampleRate, rfBandwidth, tcpPort);
 	cmd->submit(messageQueue, this);
 }
 
@@ -44,15 +53,37 @@ void TCPSrc::feed(SampleVector::const_iterator begin, SampleVector::const_iterat
 
 	if(m_spectrum != NULL)
 		m_spectrum->feed(m_sampleBuffer.begin(), m_sampleBuffer.end(), firstOfBurst);
+
+	for(int i = 0; i < m_s16leSockets.count(); i++)
+		m_s16leSockets[i].socket->write((const char*)&m_sampleBuffer[0], m_sampleBuffer.size() * 4);
+
+	if(m_s8Sockets.count() > 0) {
+		for(SampleVector::const_iterator it = m_sampleBuffer.begin(); it != m_sampleBuffer.end(); ++it) {
+			m_sampleBufferS8.push_back(it->real() >> 8);
+			m_sampleBufferS8.push_back(it->imag() >> 8);
+		}
+		for(int i = 0; i < m_s8Sockets.count(); i++)
+			m_s8Sockets[i].socket->write((const char*)&m_sampleBufferS8[0], m_sampleBuffer.size());
+	}
+
 	m_sampleBuffer.clear();
 }
 
 void TCPSrc::start()
 {
+	m_tcpServer = new QTcpServer();
+	connect(m_tcpServer, SIGNAL(newConnection()), this, SLOT(onNewConnection()));
+	m_tcpServer->listen(QHostAddress::Any, m_tcpPort);
 }
 
 void TCPSrc::stop()
 {
+	closeAllSockets(&m_s8Sockets);
+	closeAllSockets(&m_s16leSockets);
+
+	if(m_tcpServer->isListening())
+		m_tcpServer->close();
+	delete m_tcpServer;
 }
 
 bool TCPSrc::handleMessage(Message* cmd)
@@ -66,12 +97,17 @@ bool TCPSrc::handleMessage(Message* cmd)
 		m_sampleDistanceRemain = m_inputSampleRate / m_outputSampleRate;
 		cmd->completed();
 		return true;
-	} else if(cmd->id() == MsgConfigureTCPSrc::ID()) {
+	} else if(cmd->id() == MsgTCPSrcConfigure::ID()) {
-		MsgConfigureTCPSrc* cfg = (MsgConfigureTCPSrc*)cmd;
+		MsgTCPSrcConfigure* cfg = (MsgTCPSrcConfigure*)cmd;
 		m_sampleFormat = cfg->getSampleFormat();
 		m_outputSampleRate = cfg->getOutputSampleRate();
 		m_rfBandwidth = cfg->getRFBandwidth();
-		m_tcpPort = cfg->getTCPPort();
+		if(cfg->getTCPPort() != m_tcpPort) {
+			m_tcpPort = cfg->getTCPPort();
+			if(m_tcpServer->isListening())
+				m_tcpServer->close();
+			m_tcpServer->listen(QHostAddress::Any, m_tcpPort);
+		}
 		m_interpolator.create(16, m_inputSampleRate, m_rfBandwidth / 2.1);
 		m_sampleDistanceRemain = m_inputSampleRate / m_outputSampleRate;
 		cmd->completed();
@@ -80,3 +116,74 @@ bool TCPSrc::handleMessage(Message* cmd)
 		return false;
 	}
 }
+
+void TCPSrc::closeAllSockets(Sockets* sockets)
+{
+	for(int i = 0; i < sockets->count(); ++i) {
+		MsgTCPSrcConnection* msg = MsgTCPSrcConnection::create(false, sockets->at(i).id, QHostAddress(), 0);
+		msg->submit(m_uiMessageQueue, (PluginGUI*)m_tcpSrcGUI);
+		sockets->at(i).socket->close();
+	}
+}
+
+void TCPSrc::onNewConnection()
+{
+	while(m_tcpServer->hasPendingConnections()) {
+		QTcpSocket* connection = m_tcpServer->nextPendingConnection();
+		connect(connection, SIGNAL(disconnected()), this, SLOT(onDisconnected()));
+
+		switch(m_sampleFormat) {
+			case FormatS8: {
+				quint32 id = (FormatS8 << 24) | m_nextS8Id;
+				MsgTCPSrcConnection* msg = MsgTCPSrcConnection::create(true, id, connection->peerAddress(), connection->peerPort());
+				m_nextS8Id = (m_nextS8Id + 1) & 0xffffff;
+				m_s8Sockets.push_back(Socket(id, connection));
+				msg->submit(m_uiMessageQueue, (PluginGUI*)m_tcpSrcGUI);
+				break;
+			}
+
+			case FormatS16LE: {
+				quint32 id = (FormatS16LE << 24) | m_nextS16leId;
+				MsgTCPSrcConnection* msg = MsgTCPSrcConnection::create(true, id, connection->peerAddress(), connection->peerPort());
+				m_nextS16leId = (m_nextS16leId + 1) & 0xffffff;
+				m_s16leSockets.push_back(Socket(id, connection));
+				msg->submit(m_uiMessageQueue, (PluginGUI*)m_tcpSrcGUI);
+				break;
+			}
+
+			default:
+				delete connection;
+				break;
+		}
+	}
+}
+
+void TCPSrc::onDisconnected()
+{
+	quint32 id;
+	QTcpSocket* socket = NULL;
+
+	for(int i = 0; i < m_s8Sockets.count(); i++) {
+		if(m_s8Sockets[i].socket == sender()) {
+			id = m_s8Sockets[i].id;
+			socket = m_s8Sockets[i].socket;
+			m_s8Sockets.removeAt(i);
+			break;
+		}
+	}
+	if(socket == NULL) {
+		for(int i = 0; i < m_s16leSockets.count(); i++) {
+			if(m_s16leSockets[i].socket == sender()) {
+				id = m_s16leSockets[i].id;
+				socket = m_s16leSockets[i].socket;
+				m_s16leSockets.removeAt(i);
+				break;
+			}
+		}
+	}
+	if(socket != NULL) {
+		MsgTCPSrcConnection* msg = MsgTCPSrcConnection::create(false, id, QHostAddress(), 0);
+		msg->submit(m_uiMessageQueue, (PluginGUI*)m_tcpSrcGUI);
+		socket->deleteLater();
+	}
+}

plugins/demod/tcpsrc/tcpsrc.h

@@ -1,45 +1,86 @@
 #ifndef INCLUDE_TCPSRC_H
 #define INCLUDE_TCPSRC_H
 
+#include <QHostAddress>
 #include "dsp/samplesink.h"
 #include "dsp/nco.h"
 #include "dsp/interpolator.h"
 #include "util/message.h"
 
+class QTcpServer;
+class QTcpSocket;
+class TCPSrcGUI;
+
 class TCPSrc : public SampleSink {
+	Q_OBJECT
+
 public:
-	TCPSrc(SampleSink* spectrum);
+	enum SampleFormat {
+		FormatS8,
+		FormatS16LE
+	};
+
+	TCPSrc(MessageQueue* uiMessageQueue, TCPSrcGUI* tcpSrcGUI, SampleSink* spectrum);
 	~TCPSrc();
 
-	void configure(MessageQueue* messageQueue, int sampleFormat, Real outputSampleRate, Real rfBandwidth, int tcpPort);
+	void configure(MessageQueue* messageQueue, SampleFormat sampleFormat, Real outputSampleRate, Real rfBandwidth, int tcpPort);
 
 	void feed(SampleVector::const_iterator begin, SampleVector::const_iterator end, bool firstOfBurst);
 	void start();
 	void stop();
 	bool handleMessage(Message* cmd);
 
-protected:
+	class MsgTCPSrcConnection : public Message {
-	class MsgConfigureTCPSrc : public Message {
 	public:
 		static MessageRegistrator ID;
 
-		int getSampleFormat() const { return m_sampleFormat; }
+		bool getConnect() const { return m_connect; }
+		quint32 getID() const { return m_id; }
+		const QHostAddress& getPeerAddress() const { return m_peerAddress; }
+		int getPeerPort() const { return m_peerPort; }
+
+		static MsgTCPSrcConnection* create(bool connect, quint32 id, const QHostAddress& peerAddress, int peerPort)
+		{
+			return new MsgTCPSrcConnection(connect, id, peerAddress, peerPort);
+		}
+
+	private:
+		bool m_connect;
+		quint32 m_id;
+		QHostAddress m_peerAddress;
+		int m_peerPort;
+
+		MsgTCPSrcConnection(bool connect, quint32 id, const QHostAddress& peerAddress, int peerPort) :
+			Message(ID()),
+			m_connect(connect),
+			m_id(id),
+			m_peerAddress(peerAddress),
+			m_peerPort(peerPort)
+		{ }
+	};
+
+protected:
+	class MsgTCPSrcConfigure : public Message {
+	public:
+		static MessageRegistrator ID;
+
+		SampleFormat getSampleFormat() const { return m_sampleFormat; }
 		Real getOutputSampleRate() const { return m_outputSampleRate; }
 		Real getRFBandwidth() const { return m_rfBandwidth; }
 		int getTCPPort() const { return m_tcpPort; }
 
-		static MsgConfigureTCPSrc* create(int sampleFormat, Real sampleRate, Real rfBandwidth, int tcpPort)
+		static MsgTCPSrcConfigure* create(SampleFormat sampleFormat, Real sampleRate, Real rfBandwidth, int tcpPort)
 		{
-			return new MsgConfigureTCPSrc(sampleFormat, sampleRate, rfBandwidth, tcpPort);
+			return new MsgTCPSrcConfigure(sampleFormat, sampleRate, rfBandwidth, tcpPort);
 		}
 
 	private:
-		int m_sampleFormat;
+		SampleFormat m_sampleFormat;
 		Real m_outputSampleRate;
 		Real m_rfBandwidth;
 		int m_tcpPort;
 
-		MsgConfigureTCPSrc(int sampleFormat, Real outputSampleRate, Real rfBandwidth, int tcpPort) :
+		MsgTCPSrcConfigure(SampleFormat sampleFormat, Real outputSampleRate, Real rfBandwidth, int tcpPort) :
 			Message(ID()),
 			m_sampleFormat(sampleFormat),
 			m_outputSampleRate(outputSampleRate),
@@ -48,6 +89,9 @@ protected:
 		{ }
 	};
 
+	MessageQueue* m_uiMessageQueue;
+	TCPSrcGUI* m_tcpSrcGUI;
+
 	int m_inputSampleRate;
 
 	int m_sampleFormat;
@@ -60,7 +104,29 @@ protected:
 	Real m_sampleDistanceRemain;
 
 	SampleVector m_sampleBuffer;
+	std::vector<qint8> m_sampleBufferS8;
 	SampleSink* m_spectrum;
+
+	QTcpServer* m_tcpServer;
+	struct Socket {
+		quint32 id;
+		QTcpSocket* socket;
+		Socket(quint32 _id, QTcpSocket* _socket) :
+			id(_id),
+			socket(_socket)
+		{ }
+	};
+	typedef QList<Socket> Sockets;
+	Sockets m_s8Sockets;
+	Sockets m_s16leSockets;
+	quint32 m_nextS8Id;
+	quint32 m_nextS16leId;
+
+	void closeAllSockets(Sockets* sockets);
+
+protected slots:
+	void onNewConnection();
+	void onDisconnected();
 };
 
 #endif // INCLUDE_TCPSRC_H

plugins/demod/tcpsrc/tcpsrcgui.cpp

@@ -43,7 +43,16 @@ bool TCPSrcGUI::deserialize(const QByteArray& data)
 
 bool TCPSrcGUI::handleMessage(Message* message)
 {
-	return false;
+	if(message->id() == TCPSrc::MsgTCPSrcConnection::ID()) {
+		TCPSrc::MsgTCPSrcConnection* con = (TCPSrc::MsgTCPSrcConnection*)message;
+		if(con->getConnect())
+			addConnection(con->id(), con->getPeerAddress(), con->getPeerPort());
+		else delConnection(con->id());
+		message->completed();
+		return true;
+	} else {
+		return false;
+	}
 }
 
 void TCPSrcGUI::channelMarkerChanged()
@@ -57,10 +66,11 @@ TCPSrcGUI::TCPSrcGUI(PluginAPI* pluginAPI, QWidget* parent) :
 	m_pluginAPI(pluginAPI)
 {
 	ui->setupUi(this);
+	ui->connectedClientsBox->hide();
 	setAttribute(Qt::WA_DeleteOnClose, true);
 
 	m_spectrumVis = new SpectrumVis(ui->glSpectrum);
-	m_tcpSrc = new TCPSrc(m_spectrumVis);
+	m_tcpSrc = new TCPSrc(m_pluginAPI->getMainWindowMessageQueue(), this, m_spectrumVis);
 	m_channelizer = new Channelizer(m_tcpSrc);
 	m_threadedSampleSink = new ThreadedSampleSink(m_channelizer);
 	m_pluginAPI->addSampleSink(m_threadedSampleSink);
@@ -120,8 +130,21 @@ void TCPSrcGUI::applySettings()
 		outputSampleRate,
 		m_channelMarker->getCenterFrequency());
 
+	TCPSrc::SampleFormat sampleFormat;
+	switch(ui->sampleFormat->currentIndex()) {
+		case 0:
+			sampleFormat = TCPSrc::FormatS8;
+			break;
+		case 1:
+			sampleFormat = TCPSrc::FormatS16LE;
+			break;
+		default:
+			sampleFormat = TCPSrc::FormatS8;
+			break;
+	}
+
 	m_tcpSrc->configure(m_threadedSampleSink->getMessageQueue(),
-		ui->sampleFormat->currentIndex(),
+		sampleFormat,
 		outputSampleRate,
 		rfBandwidth,
 		tcpPort);
@@ -153,3 +176,22 @@ void TCPSrcGUI::on_applyBtn_clicked()
 {
 	applySettings();
 }
+
+void TCPSrcGUI::addConnection(quint32 id, const QHostAddress& peerAddress, int peerPort)
+{
+	QStringList l;
+	l.append(QString("%1:%2").arg(peerAddress.toString()).arg(peerPort));
+	new QTreeWidgetItem(ui->connections, l, id);
+	ui->connectedClientsBox->setWindowTitle(tr("Connected Clients (%1)").arg(ui->connections->topLevelItemCount()));
+}
+
+void TCPSrcGUI::delConnection(quint32 id)
+{
+	for(int i = 0; i < ui->connections->topLevelItemCount(); i++) {
+		if(ui->connections->topLevelItem(i)->type() == id) {
+			delete ui->connections->topLevelItem(i);
+			ui->connectedClientsBox->setWindowTitle(tr("Connected Clients (%1)").arg(ui->connections->topLevelItemCount()));
+			return;
+		}
+	}
+}

plugins/demod/tcpsrc/tcpsrcgui.h

@@ -1,6 +1,7 @@
 #ifndef INCLUDE_TCPSRCGUI_H
 #define INCLUDE_TCPSRCGUI_H
 
+#include <QHostAddress>
 #include "gui/rollupwidget.h"
 #include "plugin/plugingui.h"
 
@@ -54,6 +55,9 @@ private:
 	~TCPSrcGUI();
 
 	void applySettings();
+
+	void addConnection(quint32 id, const QHostAddress& peerAddress, int peerPort);
+	void delConnection(quint32 id);
 };
 
 #endif // INCLUDE_TCPSRCGUI_H

plugins/demod/tcpsrc/tcpsrcgui.ui

@@ -107,7 +107,7 @@
     </item>
    </layout>
   </widget>
-  <widget class="QWidget" name="widget_3" native="true">
+  <widget class="QWidget" name="spectrumBox" native="true">
    <property name="geometry">
     <rect>
      <x>15</x>
@@ -131,7 +131,7 @@
     </item>
    </layout>
   </widget>
-  <widget class="QWidget" name="widget_2" native="true">
+  <widget class="QWidget" name="connectedClientsBox" native="true">
    <property name="geometry">
     <rect>
      <x>15</x>
@@ -141,7 +141,7 @@
     </rect>
    </property>
    <property name="windowTitle">
-    <string>Connected Clients</string>
+    <string>Connected Clients (0)</string>
    </property>
    <layout class="QVBoxLayout" name="verticalLayout">
     <property name="spacing">
@@ -151,7 +151,7 @@
      <number>2</number>
     </property>
     <item>
-     <widget class="QTreeWidget" name="treeWidget">
+     <widget class="QTreeWidget" name="connections">
       <property name="rootIsDecorated">
        <bool>false</bool>
       </property>
@@ -191,7 +191,7 @@
   <tabstop>rfBandwidth</tabstop>
   <tabstop>tcpPort</tabstop>
   <tabstop>applyBtn</tabstop>
-  <tabstop>treeWidget</tabstop>
+  <tabstop>connections</tabstop>
  </tabstops>
  <resources/>
  <connections/>

sdrbase/dsp/threadedsamplesink.cpp

@@ -16,6 +16,8 @@ ThreadedSampleSink::ThreadedSampleSink(SampleSink* sampleSink) :
 	m_sampleFifo.moveToThread(m_thread);
 	connect(&m_sampleFifo, SIGNAL(dataReady()), this, SLOT(handleData()));
 	m_sampleFifo.setSize(262144);
+
+	sampleSink->moveToThread(m_thread);
 }
 
 ThreadedSampleSink::~ThreadedSampleSink()

sdrbase/gui/rollupwidget.cpp

@@ -222,14 +222,26 @@ void RollupWidget::mousePressEvent(QMouseEvent* event)
 	}
 }
 
+bool RollupWidget::event(QEvent* event)
+{
+	if(event->type() == QEvent::ChildAdded) {
+		((QChildEvent*)event)->child()->installEventFilter(this);
+		arrangeRollups();
+	} else if(event->type() == QEvent::ChildRemoved) {
+		((QChildEvent*)event)->child()->removeEventFilter(this);
+		arrangeRollups();
+	}
+	return QWidget::event(event);
+}
+
 bool RollupWidget::eventFilter(QObject* object, QEvent* event)
 {
 	if((event->type() == QEvent::Show) || (event->type() == QEvent::Hide)) {
 		if(children().contains(object))
 			arrangeRollups();
-	} else if((event->type() == QEvent::ChildAdded) || (event->type() == QEvent::ChildRemoved)) {
+	} else if(event->type() == QEvent::WindowTitleChange) {
-		arrangeRollups();
+		if(children().contains(object))
+			repaint();
 	}
-
 	return QWidget::eventFilter(object, event);
 }

sdrbase/mainwindow.cpp

@@ -144,6 +144,8 @@ void MainWindow::addChannelCreateAction(QAction* action)
 void MainWindow::addChannelRollup(QWidget* widget)
 {
 	((ChannelWindow*)ui->channelDock->widget())->addRollupWidget(widget);
+	ui->channelDock->show();
+	ui->channelDock->raise();
 }
 
 void MainWindow::addViewAction(QAction* action)