Adding net audio stream class

CMakeLists.txt

@@ -8,6 +8,7 @@ include( VistaCommon )
 
 # dependencies
 vista_use_package( ITABase REQUIRED FIND_DEPENDENCIES )
+vista_use_package( VistaCoreLibs REQUIRED COMPONENTS VistaInterProcComm FIND_DEPENDENCIES )
 vista_use_package( ASIO QUIET )
 vista_use_package( Portaudio QUIET )
 vista_use_package( JACK QUIET )
@@ -46,6 +47,7 @@ set( ITADataSourcesHeader
 	"include/ITADataSourcesDefinitions.h"
 	"include/ITAFileDataSink.h"
 	"include/ITAFileDataSource.h"
+	"include/ITANetAudioStream.h"
 	"include/ITAPeakDetector.h"
 	"include/ITARMSDetector.h"
 	"include/ITAStreamAmplifier.h"
@@ -66,6 +68,7 @@ set( ITADataSourcesSources
 	"src/ITADataSourceRealization.cpp"	
 	"src/ITAFileDataSink.cpp"
 	"src/ITAFileDataSource.cpp"
+	"src/ITANetAudioStream.cpp"
 	"src/ITAPeakDetector.cpp"
 	"src/ITARMSDetector.cpp"
 	"src/ITAStreamAmplifier.cpp"
@@ -131,8 +134,6 @@ endif( NOT WIN32)
 add_library( ITADataSources ${ITADataSourcesHeader} ${ITADataSourcesSources} )
 target_link_libraries( ITADataSources ${VISTA_USE_PACKAGE_LIBRARIES} )
 
-
-
 set( BUILD_SHARED_LIBS ${BUILD_SHARED_LIBS_TEMP} )
 
 

include/ITANetAudioStream.h

+/*
+ * ----------------------------------------------------------------
+ *
+ *		ITA core libs
+ *		(c) Copyright Institute of Technical Acoustics (ITA)
+ *		RWTH Aachen University, Germany, 2015-2016
+ *
+ * ----------------------------------------------------------------
+ *				    ____  __________  _______
+ *				   //  / //__   ___/ //  _   |
+ *				  //  /    //  /    //  /_|  |
+ *				 //  /    //  /    //  ___   |
+ *				//__/    //__/    //__/   |__|
+ *
+ * ----------------------------------------------------------------
+ *
+ */
+
+#ifndef INCLUDE_WATCHER_ITA_NET_AUDIO_STREAM
+#define INCLUDE_WATCHER_ITA_NET_AUDIO_STREAM
+
+#include <ITADataSourcesDefinitions.h>
+
+#include <ITADataSource.h>
+#include <ITASampleFrame.h>
+
+#include <string>
+#include <vector>
+
+class CITANetAudioStreamConnection;
+
+//! Network audio stream
+/**
+ * Audio streaming for a signal source that is connected via TCP/IP.
+ *
+ * \note not thread-safe
+ */
+class CITANetAudioStream : public ITADatasource
+{
+public:
+	CITANetAudioStream( int iChannels, double dSamplingRate, int iBufferSize, int iRingBufferCapacity );
+	virtual ~CITANetAudioStream();
+
+	bool Connect( const std::string& sAddress, int iPort );
+	bool IsConnected() const;
+	std::string GetNetworkAddress() const;
+	int GetNetworkPort() const;
+
+	int GetRingBufferSize() const;
+	
+	unsigned int GetBlocklength() const;
+	unsigned int GetNumberOfChannels() const;
+	double GetSampleRate() const;
+	const float* GetBlockPointer( unsigned int uiChannel, const ITAStreamInfo* );
+	void IncrementBlockPointer();
+
+protected:
+	int Transmit( const ITASampleFrame& sfNewSamples, int iNumSamples );
+
+private:
+	CITANetAudioStreamConnection* m_pNetAudioProducer;
+
+	double m_dSampleRate;
+	ITASampleFrame m_sfOutputStreamBuffer;
+
+	int m_iReadCursor; //!< Cursor where samples will be consumed from ring buffer on next block
+	int m_iWriteCursor; //!< Cursor where samples will feeded into ring buffer from net audio producer
+	ITASampleFrame m_sfRingBuffer;
+
+	friend class CITANetAudioStreamConnection;
+};
+
+#endif // INCLUDE_WATCHER_ITA_NET_AUDIO_STREAM

src/ITANetAudioStream.cpp

+#include <ITANetAudioStream.h>
+
+// ITA includes
+#include <ITAException.h>
+
+// Vista includes
+#include <VistaInterProcComm/Concurrency/VistaThreadLoop.h>
+#include <VistaInterProcComm/Connections/VistaConnectionIP.h>
+#include <VistaInterProcComm/IPNet/VistaTCPServer.h>
+#include <VistaInterProcComm/IPNet/VistaTCPSocket.h>
+//#include <VistaBase/VistaTimeUtils.h>
+#include <VistaInterProcComm/IPNet/VistaIPAddress.h>
+
+// STL
+#include <cmath>
+
+class CITANetAudioStreamConnection : public VistaThreadLoop
+{
+public:
+	enum MessageType
+	{
+		NET_MESSAGE_NONE = 0,
+		NET_MESSAGE_OPEN,
+		NET_MESSAGE_CLOSE,
+		NET_MESSAGE_SAMPLES,
+	};
+
+	inline CITANetAudioStreamConnection( CITANetAudioStream* pParent )
+		: m_pParent( pParent )
+		, m_pConnection( NULL )
+		, m_bStopIndicated( false )
+	{
+	};
+
+	inline bool Connect( const std::string& sAddress, int iPort )
+	{
+		if( m_pConnection )
+			ITA_EXCEPT1( MODAL_EXCEPTION, "This net stream is already connected" );
+
+		// Attempt to connect and check parameters
+		m_pConnection = new VistaConnectionIP( VistaConnectionIP::CT_TCP, sAddress, iPort );
+		if( !m_pConnection->GetIsConnected() )
+		{
+			delete m_pConnection;
+			m_pConnection = NULL;
+			return false;
+		}
+
+		int iMessageType = NET_MESSAGE_OPEN;
+		m_pConnection->Send( &iMessageType, sizeof( int ) );
+
+		int iNumChannels = ( int ) m_pParent->GetNumberOfChannels();
+		m_pConnection->Send( &iNumChannels, sizeof( int ) );
+		double dSampleRate = m_pParent->GetSampleRate();
+		m_pConnection->Send( &dSampleRate, sizeof( double ) );
+		int iBlockLength = ( int ) m_pParent->GetBlocklength();
+		m_pConnection->Send( &iBlockLength, sizeof( int ) );
+		int iRingBufferSize = ( int ) m_pParent->GetRingBufferSize();
+		m_pConnection->Send( &iRingBufferSize, sizeof( int ) );
+		m_pConnection->WaitForSendFinish();
+
+		int iServerMessageType;
+		m_pConnection->Receive( &iServerMessageType, sizeof( int ) );
+
+		Run();
+	};
+
+	inline void Disconnect()
+	{
+		m_bStopIndicated = true;
+		StopGently( true );
+
+		delete m_pConnection;
+		m_pConnection = NULL;
+
+		m_bStopIndicated = false;
+	};
+
+	inline ~CITANetAudioStreamConnection()
+	{
+		int iMessageType = NET_MESSAGE_CLOSE;
+		m_pConnection->Send( &iMessageType, sizeof( int ) );
+	};
+
+	inline bool LoopBody()
+	{
+		if( m_bStopIndicated )
+			return true;
+
+		// Receive messages
+		while( true )
+		{
+			m_pConnection->Receive( NULL, 0 ); // @todo: receive messages and react
+
+			int iNumSamples = 12;
+			if( true )
+				m_pParent->Transmit( m_sfReceivingBuffer, iNumSamples );
+		}
+	};
+
+private:
+	VistaConnectionIP* m_pConnection;
+	CITANetAudioStream* m_pParent;
+	ITASampleFrame m_sfReceivingBuffer;	
+	bool m_bStopIndicated;
+};
+
+CITANetAudioStream::CITANetAudioStream( int iChannels, double dSamplingRate, int iBufferSize, int iRingBufferCapacity )
+	: m_sfOutputStreamBuffer( iChannels, iBufferSize, true )
+	, m_dSampleRate( dSamplingRate )
+	, m_sfRingBuffer( iChannels, iRingBufferCapacity, true )
+	
+{
+	m_pNetAudioProducer = new CITANetAudioStreamConnection( this );
+}
+
+bool CITANetAudioStream::Connect( const std::string& sAddress, int iPort )
+{
+	return m_pNetAudioProducer->Connect( sAddress, iPort );
+}
+
+CITANetAudioStream::~CITANetAudioStream()
+{
+	delete m_pNetAudioProducer;
+}
+
+int CITANetAudioStream::GetRingBufferSize() const
+{
+	return m_sfRingBuffer.GetLength();
+}
+
+unsigned int CITANetAudioStream::GetBlocklength() const
+{
+	return  ( unsigned int ) m_sfOutputStreamBuffer.GetLength();
+}
+
+unsigned int CITANetAudioStream::GetNumberOfChannels() const
+{
+	return ( unsigned int ) m_sfOutputStreamBuffer.channels();
+}
+
+double CITANetAudioStream::GetSampleRate() const
+{
+	return m_dSampleRate;
+}
+
+const float* CITANetAudioStream::GetBlockPointer( unsigned int uiChannel, const ITAStreamInfo* )
+{
+	// @todo: implement cyclic read from ring buffer
+	
+	return m_sfOutputStreamBuffer[ uiChannel ].GetData();
+}
+
+void CITANetAudioStream::IncrementBlockPointer()
+{
+	// Increment read cursor by one audio block and wrap around if exceeding ring buffer
+	m_iReadCursor = ( m_iReadCursor + m_sfOutputStreamBuffer.GetLength() ) % m_sfRingBuffer.GetLength();
+}
+
+int CITANetAudioStream::Transmit( const ITASampleFrame& sfNewSamples, int iNumSamples )
+{
+	ITA_EXCEPT0( NOT_IMPLEMENTED );
+}