merge

CMakeLists.txt

@@ -52,6 +52,7 @@ set( ITADataSourcesHeader
 	"include/ITANetAudioProtocol.h"
 	"include/ITANetAudioServer.h"
 	"include/ITANetAudioStream.h"
+	"include/ITANetAudioStreamingClient.h"
 	"include/ITANetAudioStreamingServer.h"
 	"include/ITANetAudioStreamingClient.h"
 	"include/ITAPeakDetector.h"
@@ -77,6 +78,7 @@ set( ITADataSourcesSources
 	"src/ITANetAudioMessage.cpp"
 	"src/ITANetAudioProtocol.cpp"
 	"src/ITANetAudioStream.cpp"
+	"src/ITANetAudioStreamingClient.cpp"
 	"src/ITANetAudioStreamingServer.cpp"
 	"src/ITANetAudioServer.cpp"
 	"src/ITAPeakDetector.cpp"

include/ITANetAudioClient.h

@@ -16,51 +16,39 @@
  *
  */
 
-#ifndef INCLUDE_WATCHER_ITA_NET_AUDIO_CONNECTION
-#define INCLUDE_WATCHER_ITA_NET_AUDIO_CONNECTION
+#ifndef INCLUDE_WATCHER_ITA_NET_AUDIO_CLIENT
+#define INCLUDE_WATCHER_ITA_NET_AUDIO_CLIENT
 
 #include <ITADataSourcesDefinitions.h>
 
-#include <ITADataSource.h>
-#include <ITASampleFrame.h>
-
-#include <VistaInterProcComm/Concurrency/VistaThreadLoop.h>
-
 #include <string>
-#include <vector>
-
-class CITANetAudioMessage;
-class CITANetAudioProtocol;
-class CITANetAudioStream;
 
 class VistaConnectionIP;
 
-class CITANetAudioClient : public VistaThreadLoop
+//! A network audio client that connects to a network audio server
+/** 
+  * Use CITANetAudioStreamingClient to start an audio stream with the connection of this client.
+  *
+  */
+class CITANetAudioClient
 {
 public:
 
-	//! Create an network audio client that feeds into a network audio stream
+	//! Create an network audio client that connects to a network audio server
 	/**
 	  * \param pParent ITADataSource-compatible audio stream
 	  */
-	CITANetAudioClient( CITANetAudioStream* pParent );
+	CITANetAudioClient();
 	~CITANetAudioClient();
 
 	bool Connect( const std::string& sAddress, int iPort );
 	void Disconnect();
 	bool GetIsConnected() const;
 
-	bool LoopBody();
+	VistaConnectionIP* GetConnection() const;
 
 private:
-	CITANetAudioStream* m_pParent;
-
 	VistaConnectionIP* m_pConnection;
-	CITANetAudioProtocol* m_pProtocol;
-	CITANetAudioMessage* m_pMessage;
-
-	ITASampleFrame m_sfReceivingBuffer;
-	bool m_bStopIndicated;
 };
 
-#endif // INCLUDE_WATCHER_ITA_NET_AUDIO_CONNECTION
+#endif // INCLUDE_WATCHER_ITA_NET_AUDIO_CLIENT

include/ITANetAudioMessage.h

@@ -25,6 +25,7 @@
 #include <ITAException.h>
 #include <ITASampleBuffer.h>
 #include <ITASampleFrame.h>
+#include <ITANetAudioProtocol.h>
 
 // Vista includes
 #include <VistaInterProcComm/Connections/VistaByteBufferSerializer.h>
@@ -77,6 +78,7 @@ public:
 	void WriteString( const std::string& );
 	void WriteIntVector( const std::vector< int > );
 	void WriteFloatVector( const std::vector< float > );
+	void WriteStreamingParameters( const CITANetAudioProtocol::StreamingParameters &);
 
 	std::string ReadString();
 	int ReadInt();
@@ -86,8 +88,8 @@ public:
 	double ReadDouble();
 	std::vector< int > ReadIntVector();
 	std::vector< float > ReadFloatVector();
+	CITANetAudioProtocol::StreamingParameters ReadStreamingParameters();
 
-	void WriteClientOpen();
 
 private:
 	int m_nMessageType;

include/ITANetAudioProtocol.h

@@ -63,6 +63,26 @@ public:
 
 	CITANetAudioProtocol();
 	virtual ~CITANetAudioProtocol();
+
+	struct StreamingParameters
+	{
+		int iChannels;
+		double dSampleRate;
+		int iBlockSize;
+
+		inline StreamingParameters()
+		{
+			iChannels = 0;
+			dSampleRate = 0.0f;
+			iBlockSize = 0;
+		};
+
+		inline bool operator==( const StreamingParameters& rhs )
+		{
+			if( ( iChannels == rhs.iChannels ) && ( dSampleRate == rhs.dSampleRate ) && ( iBlockSize == rhs.iBlockSize ) )
+				return true;
+		}
+	};
 };
 
 #endif // INCLUDE_WATCHER_ITA_NET_AUDIO_PROTOCOL

include/ITANetAudioStream.h

@@ -27,7 +27,7 @@
 #include <string>
 #include <vector>
 
-class CITANetAudioClient;
+class CITANetAudioStreamingClient;
 
 //! Network audio stream
 /**
@@ -38,7 +38,7 @@ class CITANetAudioClient;
 class ITA_DATA_SOURCES_API CITANetAudioStream : public ITADatasource
 {
 public:
-	CITANetAudioStream( int iChannels, double dSamplingRate, int iBufferSize, int iRingBufferCapacity );
+	CITANetAudioStream( int iChannels, double dSamplingRate, int iBufferSize, int iRingBufferCapacity = 2048 );
 	virtual ~CITANetAudioStream();
 
 	bool Connect( const std::string& sAddress, int iPort );
@@ -53,10 +53,19 @@ public:
 	void IncrementBlockPointer();
 
 protected:
+	//! This method is called by the streaming client and pushes sampes into the ring buffer
+	/**
+	  * \param sfNewSamples Sample buffer (multi channel) with sample data
+	  * \param iNumSamples samples to be read from the sample frame (must be smaller or equal length)
+	  *
+	  * \return Number of free samples in ring buffer
+	  */
 	int Transmit( const ITASampleFrame& sfNewSamples, int iNumSamples );
 
+	int GetRingbufferFreeSamples();
+
 private:
-	CITANetAudioClient* m_pNetAudioProducer;
+	CITANetAudioStreamingClient* m_pNetAudioStreamingClient;
 
 	double m_dSampleRate;
 	ITASampleFrame m_sfOutputStreamBuffer;
@@ -65,7 +74,7 @@ private:
 	int m_iWriteCursor; //!< Cursor where samples will feeded into ring buffer from net audio producer
 	ITASampleFrame m_sfRingBuffer; //!< Buffer incoming data
 
-	friend class CITANetAudioClient;
+	friend class CITANetAudioStreamingClient;
 };
 
 #endif // INCLUDE_WATCHER_ITA_NET_AUDIO_STREAM

include/ITANetAudioStreamingClient.h

@@ -21,30 +21,36 @@
 
 #include <ITADataSourcesDefinitions.h>
 
-#include <VistaInterProcComm/Concurrency/VistaThreadLoop.h>
-#include <ITADataSource.h>
+#include <ITANetAudioProtocol.h>
+
 #include <ITASampleFrame.h>
 
+#include <VistaInterProcComm/Concurrency/VistaThreadLoop.h>
+
 #include <string>
 #include <vector>
 
 class CITANetAudioClient;
+class CITANetAudioMessage;
+class CITANetAudioProtocol;
 class CITANetAudioStream;
 
 //! Network audio streaming client
 /**
  * Audio streaming for a signal source that is connected via TCP/IP.
+ * Implements the ITA network protocol for audio streaming in client side.
  *
  * \note not thread-safe
  */
 class ITA_DATA_SOURCES_API CITANetAudioStreamingClient : public VistaThreadLoop
 {
 public:
-	CITANetAudioStreamingClient( int iChannels, double dSamplingRate, int iBufferSize, int iRingBufferCapacity );
+	CITANetAudioStreamingClient( CITANetAudioStream* pParent );
 	virtual ~CITANetAudioStreamingClient();
 
 	bool Connect( const std::string& sAddress, int iPort );
 	bool GetIsConnected() const;
+	void Disconnect();
 
 	bool LoopBody();
 
@@ -52,8 +58,17 @@ private:
 	CITANetAudioClient* m_pClient;
 	CITANetAudioStream* m_pStream;
 
+	CITANetAudioProtocol* m_pProtocol;
+	CITANetAudioMessage* m_pMessage;
+	VistaConnectionIP* m_pConnection;
+
 	ITASampleFrame m_sfReceivingBuffer; //!< Buffer incoming data
 
+	CITANetAudioProtocol::StreamingParameters m_oClientParams;
+	CITANetAudioProtocol::StreamingParameters m_oServerParams;
+
+	bool m_bStopIndicated;
+
 	friend class CITANetAudioClient;
 };
 

src/ITANetAudioClient.cpp

@@ -6,91 +6,45 @@
 
 #include <VistaInterProcComm/Connections/VistaConnectionIP.h>
 
-CITANetAudioClient::CITANetAudioClient( CITANetAudioStream* pParent )
-	: m_pParent( pParent )
-	, m_pConnection( NULL )
-	, m_bStopIndicated( false )
+CITANetAudioClient::CITANetAudioClient()
+	: m_pConnection( NULL )
 {
-	m_pMessage = new CITANetAudioMessage( VistaSerializingToolset::SWAPS_MULTIBYTE_VALUES );
 }
 
 CITANetAudioClient::~CITANetAudioClient()
 {
-	if( m_pConnection )
-	{
-		m_pMessage->ResetMessage();
-		m_pMessage->SetMessageType( CITANetAudioProtocol::NP_CLIENT_CLOSE );
-		m_pMessage->WriteAnswer();
-	}
-
-	delete m_pMessage;
+	delete m_pConnection;
 }
 
 bool CITANetAudioClient::Connect( const std::string& sAddress, int iPort )
 {
-	if( m_pConnection )
+	if( GetIsConnected() )
 		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() )
+	if( !GetIsConnected() )
 	{
 		delete m_pConnection;
 		m_pConnection = NULL;
 		return false;
 	}
-	
-	m_pMessage->SetConnection( m_pConnection );
-
-	m_pMessage->ResetMessage();
-	m_pMessage->WriteClientOpen();
-	m_pMessage->WriteMessage();
-	m_pMessage->ReadAnswer();
-
-	Run();
 
 	return true;
 }
 
-void CITANetAudioClient::Disconnect()
+VistaConnectionIP* CITANetAudioClient::GetConnection() const
 {
-	m_bStopIndicated = true;
-	StopGently( true );
-
-	delete m_pConnection;
-	m_pConnection = NULL;
-
-	m_bStopIndicated = false;
+	return m_pConnection;
 }
 
-bool CITANetAudioClient::LoopBody()
+void CITANetAudioClient::Disconnect()
 {
-	if( m_bStopIndicated )
-		return true;
-
-	// Receive message
-	m_pMessage->ReadMessage();
-	switch( m_pMessage->GetMessageType() )
-	{
-	case CITANetAudioProtocol::NP_INVALID:
-		break;
-	case CITANetAudioProtocol::NP_SERVER_SEND_SAMPLES:
-		/*
-		int iNumSamples = m_pMessage->ReadSamples( m_sfReceivingBuffer );
-		m_pParent->Transmit( m_sfReceivingBuffer, iNumSamples );
-		int iFreeSamples = m_pParent->GetRingBufferFreeSamples();
-		m_pMessage->WriteFreeRingBufferSamples( iFreeSamples );
-		m_pMessage->WriteAnswer();
-		*/
-		break;
-	}
+	delete m_pConnection;
+	m_pConnection = NULL;
 }
 
-
 bool CITANetAudioClient::GetIsConnected() const
 {
-	if( m_pConnection )
-		return true;
-	else
-		return false;
+	return m_pConnection ? true : false;
 }

src/ITANetAudioMessage.cpp

 #include <ITANetAudioMessage.h>
-#include <ITANetAudioProtocol.h>
 
 #include <ITAStringUtils.h>
 
@@ -377,7 +376,20 @@ std::vector<int> CITANetAudioMessage::ReadIntVector()
 	return viData;
 }
 
-void CITANetAudioMessage::WriteClientOpen()
+CITANetAudioProtocol::StreamingParameters CITANetAudioMessage::ReadStreamingParameters()
 {
-	SetMessageType( CITANetAudioProtocol::NP_CLIENT_OPEN );
+	CITANetAudioProtocol::StreamingParameters oParams;
+
+	oParams.iChannels = ReadInt();
+	oParams.dSampleRate = ReadDouble();
+	oParams.iBlockSize = ReadInt();
+
+	return oParams;
+}
+
+void CITANetAudioMessage::WriteStreamingParameters( const CITANetAudioProtocol::StreamingParameters & oParams )
+{
+	WriteInt( oParams.iChannels );
+	WriteDouble( oParams.dSampleRate );
+	WriteInt( oParams.iBlockSize );
 }

src/ITANetAudioStream.cpp

 #include <ITANetAudioStream.h>
 
+#include <ITANetAudioStreamingClient.h>
+
 // ITA includes
 #include <ITAException.h>
-#include <ITANetAudioMessage.h>
-#include <ITANetAudioProtocol.h>
-#include <ITANetAudioClient.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>
@@ -23,24 +14,27 @@ CITANetAudioStream::CITANetAudioStream( int iChannels, double dSamplingRate, int
 	, m_sfRingBuffer( iChannels, iRingBufferCapacity, true )
 	
 {
-	m_pNetAudioProducer = new CITANetAudioClient( this );
+	if( iBufferSize > iRingBufferCapacity )
+		ITA_EXCEPT1( INVALID_PARAMETER, "Ring buffer capacity can not be smaller than buffer size." );
+
+	m_pNetAudioStreamingClient = new CITANetAudioStreamingClient( this );
 	m_iReadCursor = 0;
 	m_iWriteCursor = 0;
 }
 
 CITANetAudioStream::~CITANetAudioStream()
 {
-	delete m_pNetAudioProducer;
+	delete m_pNetAudioStreamingClient;
 }
 
 bool CITANetAudioStream::Connect( const std::string& sAddress, int iPort )
 {
-	return m_pNetAudioProducer->Connect( sAddress, iPort );
+	return m_pNetAudioStreamingClient->Connect( sAddress, iPort );
 }
 
 bool CITANetAudioStream::GetIsConnected() const
 {
-	return m_pNetAudioProducer->GetIsConnected();
+	return m_pNetAudioStreamingClient->GetIsConnected();
 }
 
 const float* CITANetAudioStream::GetBlockPointer( unsigned int uiChannel, const ITAStreamInfo* )
@@ -84,6 +78,11 @@ int CITANetAudioStream::Transmit( const ITASampleFrame& sfNewSamples, int iNumSa
 	}	
 }
 
+int CITANetAudioStream::GetRingbufferFreeSamples()
+{
+	ITA_EXCEPT0( NOT_IMPLEMENTED );
+}
+
 int CITANetAudioStream::GetRingBufferSize() const
 {
 	return m_sfRingBuffer.GetLength();

src/ITANetAudioStreamingClient.cpp

+#include <ITANetAudioStreamingClient.h>
+
+#include <ITANetAudioClient.h>
+#include <ITANetAudioMessage.h>
+#include <ITANetAudioStream.h>
+
+#include <VistaInterProcComm/Connections/VistaConnectionIP.h>
+
+CITANetAudioStreamingClient::CITANetAudioStreamingClient( CITANetAudioStream* pParent )
+	: m_pStream( pParent )
+	, m_pConnection( NULL )
+	, m_bStopIndicated( false )
+{
+	m_pClient = new CITANetAudioClient();
+
+	m_oClientParams.iChannels = pParent->GetNumberOfChannels();
+	m_oClientParams.dSampleRate = pParent->GetSampleRate();
+	m_oClientParams.iBlockSize = pParent->GetBlocklength();
+}
+
+CITANetAudioStreamingClient::~CITANetAudioStreamingClient()
+{
+	if( m_pConnection )
+	{
+		m_pMessage->ResetMessage();
+		m_pMessage->SetMessageType( CITANetAudioProtocol::NP_CLIENT_CLOSE );
+		m_pMessage->WriteAnswer();
+	}
+}
+
+bool CITANetAudioStreamingClient::Connect( const std::string& sAddress, int iPort )
+{
+	if( GetIsConnected() )
+		return false;
+	
+	if( !m_pClient->Connect( sAddress, iPort ) )
+		ITA_EXCEPT1( INVALID_PARAMETER, "Could not connect to " + sAddress );
+	
+	m_pConnection = m_pClient->GetConnection();
+
+	m_pMessage->SetConnection( m_pConnection );
+
+	// Validate streaming parameters of server and client
+	m_pMessage->ResetMessage();
+	m_pMessage->SetMessageType( CITANetAudioProtocol::NP_CLIENT_OPEN );
+	m_pMessage->WriteStreamingParameters( m_oClientParams );
+	m_pMessage->WriteMessage();
+
+	m_pMessage->ReadAnswer();
+	CITANetAudioProtocol::StreamingParameters oServerParams = m_pMessage->ReadStreamingParameters();
+	if( oServerParams == m_oClientParams )
+		m_oServerParams = oServerParams;
+	else
+		ITA_EXCEPT1( INVALID_PARAMETER, "Streaming parameters of network audio server and client do not match." );
+
+	Run();
+
+	return true;
+}
+
+void CITANetAudioStreamingClient::Disconnect()
+{
+	m_bStopIndicated = true;
+	StopGently( true );
+
+	delete m_pConnection;
+	m_pConnection = NULL;
+
+	m_bStopIndicated = false;
+}
+
+bool CITANetAudioStreamingClient::LoopBody()
+{
+	if( m_bStopIndicated )
+		return true;
+
+	// Receive message
+	m_pMessage->ReadMessage();
+	switch( m_pMessage->GetMessageType() )
+	{
+	case CITANetAudioProtocol::NP_INVALID:
+		break;
+	case CITANetAudioProtocol::NP_SERVER_SEND_SAMPLES:
+		/*
+		int iNumSamples = m_pMessage->ReadSamples( m_sfReceivingBuffer );
+		m_pParent->Transmit( m_sfReceivingBuffer, iNumSamples );
+		int iFreeSamples = m_pParent->GetRingBufferFreeSamples();
+		m_pMessage->WriteFreeRingBufferSamples( iFreeSamples );
+		m_pMessage->WriteAnswer();
+		*/
+		break;
+	}
+}
+
+bool CITANetAudioStreamingClient::GetIsConnected() const
+{
+	return m_pClient->GetIsConnected();
+}