CMakeLists.txt

@@ -110,23 +110,22 @@ endif( )
 
 if( ITA_DATA_SOURCES_WITH_NET_AUDIO )
 	list( APPEND ITADataSourcesHeader
-		"include/ITANetAudioClient.h"
-		"include/ITANetAudioMessage.h"
-		"include/ITANetAudioProtocol.h"
-		"include/ITANetAudioServer.h"
 		"include/ITANetAudioStream.h"
-		"include/ITANetAudioStreamingClient.h"
+		"include/ITANetAudioSampleServer.h"
 		"include/ITANetAudioStreamingServer.h"
-		"include/ITANetAudioStreamingClient.h"
 		)
 	list( APPEND ITADataSourcesSources
 		"src/ITANetAudioClient.cpp"
+		"src/ITANetAudioClient.h"
 		"src/ITANetAudioMessage.cpp"
-		"src/ITANetAudioProtocol.cpp"
+		"src/ITANetAudioMessage.h"
+		"src/ITANetAudioProtocol.h"
+		"src/ITANetAudioServer.cpp"
+		"src/ITANetAudioServer.h"
 		"src/ITANetAudioStream.cpp"
 		"src/ITANetAudioStreamingClient.cpp"
+		"src/ITANetAudioStreamingClient.h"
 		"src/ITANetAudioStreamingServer.cpp"
-		"src/ITANetAudioServer.cpp"
 		)	
 endif( )
 

include/ITANetAudioSampleServer.h

+/*
+ * ----------------------------------------------------------------
+ *
+ *		ITA core libs
+ *		(c) Copyright Institute of Technical Acoustics (ITA)
+ *		RWTH Aachen University, Germany, 2015-2017
+ *
+ * ----------------------------------------------------------------
+ *				    ____  __________  _______
+ *				   //  / //__   ___/ //  _   |
+ *				  //  /    //  /    //  /_|  |
+ *				 //  /    //  /    //  ___   |
+ *				//__/    //__/    //__/   |__|
+ *
+ * ----------------------------------------------------------------
+ *
+ */
+
+#ifndef INCLUDE_WATCHER_ITA_NET_AUDIO_SAMPLE_SERVER
+#define INCLUDE_WATCHER_ITA_NET_AUDIO_SAMPLE_SERVER
+
+#include <ITADataSourcesDefinitions.h>
+#include <ITANetAudioStreamingServer.h>
+#include <ITADataSourceRealization.h>
+
+
+//! Sample-generation class with abstract method for providing samples
+/*
+ * This ready-to-use class helps to provide samples for a NetAudio streaming server with
+ * a single method for processing that has to be implemented ...
+ * ... just derive and implement Process() method. Have a look at Zero() method
+ * for exemplary usage of sample buffer.
+ */ 
+class CITASampleProcessor : public ITADatasourceRealization
+{
+public:
+	//! Create a sample processor with streaming parameters
+	/*
+	 * @param[in] iNumChannels Channels provided
+	 * @param[in] dSampleRate Audio processing sampling rate
+	 * @param[in] iBlockLength Audio processing block length / buffer size
+	 */ 
+	inline CITASampleProcessor( const int iNumChannels, const double dSampleRate, const int iBlockLength )
+		: ITADatasourceRealization( ( unsigned int ) ( iNumChannels ), dSampleRate, ( unsigned int ) ( iBlockLength ) )
+	{
+		m_vvfSampleBuffer.resize( iNumChannels );
+		for( size_t c = 0; c < iNumChannels; c++ )
+			m_vvfSampleBuffer[ c ].resize( iBlockLength );
+
+		Zero();
+	};
+
+	inline ~CITASampleProcessor()
+	{
+	};
+
+	//! Sets all channels and samples to zero
+	inline void Zero()
+	{
+		/* 
+		 * Use this as an example how to work with the buffer structure.
+		*/
+		
+		// Iterate over channels
+		for( size_t c = 0; c < m_vvfSampleBuffer.size(); c++ ) 
+		{
+			std::vector< float >& vfSingleChannelSampleBuffer( m_vvfSampleBuffer[ c ] ); // One channel
+			
+			// Iterate over samples of channel
+			for( size_t n = 0; n < vfSingleChannelSampleBuffer.size(); n++ )
+			{
+				float& fSample( vfSingleChannelSampleBuffer[ n ] ); // One sample
+				fSample = 0.0f; // -> Manipulation
+			}
+		}
+	};
+	
+	//! Process samples (overwrite this virtual method)
+	/**
+	  * Method that is called in audio streaming context and requests
+	  * to produce or copy audio samples into the internal buffer m_vvfSampleBuffer
+	  *
+	  * @param[in] pStreamInfo Information over streaming status, i.e. sample count and time stamp
+	  *
+	  */
+	virtual void Process( const ITAStreamInfo* pStreamInfo ) =0;
+
+protected:
+	std::vector< std::vector< float > > m_vvfSampleBuffer; //!< Multi-channel sample buffer to be filled
+
+private:
+	//! Delegate internal buffer to audio stream (ITADatasource)
+	inline void ProcessStream( const ITAStreamInfo* pInfo )
+	{
+		Process( pInfo );
+
+		for( size_t c = 0; c < m_vvfSampleBuffer.size(); c++ )
+		{
+			float* pfData = GetWritePointer( ( unsigned int ) ( c ) );
+			for( size_t n = 0; n < m_vvfSampleBuffer[ c ].size(); n++ )
+				pfData[ n ] = m_vvfSampleBuffer[ c ][ n ];
+		}
+
+		IncrementWritePointer();
+	};
+};
+
+//! Network audio sample server (for providing samples via derived generator class)
+/**
+  * Audio sample transmitter for a networked sample callback function that can connect via TCP/IP.
+  *
+  * @sa CITANetAudioStream CITANetAudioStreamingServer CITASampleProcessor
+  * @note not thread-safe
+  */
+class CITANetAudioSampleServer : public CITANetAudioStreamingServer
+{
+public:
+	inline CITANetAudioSampleServer( CITASampleProcessor* pProcessor )
+		: m_pSampleProcessor( pProcessor )
+	{
+		SetInputStream( m_pSampleProcessor );
+	};
+
+	inline ~CITANetAudioSampleServer()
+	{};
+
+private:
+	//! Prohibit public access to streaming context and delegate
+	inline void SetInputStream( ITADatasource* pDataSource )
+	{
+		CITANetAudioStreamingServer::SetInputStream( pDataSource );
+	};
+
+	//! Prohibit public access to streaming context and delegate
+	inline ITADatasource* GetInputStream() const
+	{
+		return CITANetAudioStreamingServer::GetInputStream();
+	};
+
+	CITASampleProcessor* m_pSampleProcessor; //!< Callback / sample processor
+};
+
+#endif // INCLUDE_WATCHER_ITA_NET_AUDIO_SAMPLE_SERVER

include/ITANetAudioStream.h

@@ -26,9 +26,6 @@
 
 #include <string>
 #include <vector>
-#include <iostream>
-#include <fstream>
-using namespace std;
 
 
 class CITANetAudioStreamingClient;
@@ -38,7 +35,7 @@ class ITABufferedDataLoggerImplAudio;
 
 //! Network audio stream
 /**
-  * Audio streaming for a signal source that is connected via TCP/IP.
+  * Audio streaming for a signal source that is connected via TCP/IP or UDP.
   * The network audio stream behaves like a client and receives samples
   * from a network audio stream server, CITANetAudioStreamingSearver.
   * 
@@ -46,11 +43,13 @@ class ITABufferedDataLoggerImplAudio;
   * block the streaming processing, because it is decoupled from the 
   * network connection and forwards samples from an internal ring buffer.
   * If the buffer runs out of samples, zeros will be return. If the buffer
-  * overruns, the sample server will be suspendeb by blocking the network
+  * overruns, the sample server will be suspended by blocking the network
   * data flow.
   *
-  * Latency can be managed by either providing a small ring buffer or
-  * oversizing the ring buffer and requesting a target latency.
+  * Latency can be managed by either providing a small ring buffer and 
+  * constantly filling it uo, or by oversizing the internal ring buffer 
+  * only pushing samples to meet a target latency. This has to be 
+  * implemented by the server.
   *
   * \note not thread-safe
   */
@@ -58,7 +57,21 @@ class ITA_DATA_SOURCES_API CITANetAudioStream : public ITADatasource
 {
 public:
 	//! Constructor of a network audio stream
-	CITANetAudioStream( int iChannels, double dSamplingRate, int iBufferSize, int iRingBufferCapacity = 2048 );
+	/**
+	  * @param[in] iChannels Number of channels
+	  * @param[in] dSamplingRate Sampling rate
+	  * @param[in] iBufferSize Size of audio streaming buffer
+	  * @param[in] iRingBufferCapacity Internal ring buffer
+	  *
+	  * The ring buffer capacity should be roughly 6-10 buffer sizes long for short audio streaming buffers,
+	  * and can go down to one block in case of higher audio buffer sizes.
+	  *
+	  * The streaming parameters have to match with the server settings (yes also buffer size, that of the audio streaming context)
+	  *
+	  * @note Accept for more memory usage, oversizing the buffer does not require more CPU.
+	  */
+	CITANetAudioStream( const int iChannels, const double dSamplingRate, const int iBufferSize, const int iRingBufferCapacity = 2048 );
+
 	virtual ~CITANetAudioStream();
 
 	//! Network streaming status of client
@@ -68,16 +81,20 @@ public:
 		STOPPED, //!< Client not connected to a server and streaming stopped, i.e. not receiving samples by choice
 		CONNECTED, //!< Client is connected to a sample server (and potentially receives samples)
 		STREAMING, //!< 
-		BUFFER_UNDERRUN,
-		BUFFER_OVERRUN,
+		BUFFER_UNDERRUN, //!< Client internal audio buffer ran out of samples
+		BUFFER_OVERRUN, //!< Client internal audio ring buffer is full
 	};
 
 	//! Connect a streaming server
 	/**
-	  * @sAddress[in] Server address IP, i.e. 127.0.0.1
+	  * @sAddress[in] Server address IP (127.0.0.1, localhost, etc.)
 	  * @iPort[in] Server socket port, defaults to 12480
+	  * @return True, if connection could be established and streaming parameters match
 	  */
-	bool Connect( const std::string& sAddress, int iPort = 12480 );
+	bool Connect( const std::string& sAddress, const int iPort = 12480, const bool bUseUDP = false );
+
+	//! Disconnct safely from server
+	void Disconnect();
 
 	//! Returns the connection status
 	/**
@@ -85,43 +102,47 @@ public:
 	  */
 	bool GetIsConnected() const;
 
-	//! Set allowed latency (s)
+	//! Returns the minimal latency possible (single block)
 	/**
-	  * Sets the latency that will be used for reading and writing from ring buffer.
-	  * New samples will be requested and send if the latency / ring buffer samples
-	  * is lower than the target latency.
-	  */
-	void SetAllowedLatencySeconds( float fLatencySeconds );
-	void SetAllowedLatencySamples( int iLatencySamples );
-	float GetAllowedLatencySeconds() const;
-	int GetAllowedLatencySamples() const;
-	
-	//! Sets the minimal latency possible
-	/**
-	  * Real-time network audio is considered to process at lowest latency possible.
-	  * However, this implementation requires at least one block. Hence latency is
-	  * depending on sampling rate and block length.
-	  *
-	  * @sa GetMinimumLatencySamples()
-	  * @sa GetMinimumLatencySamples()
+	  * @return Minimum latency in seconds
 	  */
 	float GetMinimumLatencySeconds() const;
 
+	//! Returns the maximum latency possible (entire ring buffer used)
+	/**
+	  * @return Maximum latency in seconds
+	  */
 	float GetMaximumLatencySeconds() const;
+	
+	//! Returns the minimum latency possible (single block)
+	/**
+	  * @return Minimum latency in samples
+	  */
 	int GetMinimumLatencySamples() const;
+
+	//! Returns the maximum latency possible (entire ring buffer used)
+	/**
+	  * @return Maximum latency in samples
+	  */
 	int GetMaximumLatencySamples() const;
 
-	//! Sets the latency for real-time processing
+	//! Returns the NetAudio streaming logger base name
+	std::string GetNetAudioStreamLoggerBaseName() const;
+
+	//! Sets the NetAudio streaming logger base name
 	/**
-	  * Real-time network audio is considered to process at lowest latency possible.
-	  * However, this implementation requires at least one block. Hence latency is
-	  * depending on sampling rate and block length. This method basically
-	  * sets the minimum allowed latency to this value.
-	  *
-	  * @sa GetMinimumLatencySeconds()
-	  * @sa SetAllowedLatencySeconds()
+	  * If debugging is enabled, all debugging files will be named
+	  * with this suffix.
+	  * @param[in] sBaseName Base name string
+	  * 
 	  */
-	void SetLatencyForRealtime();
+	void SetNetAudioStreamingLoggerBaseName( const std::string& sBaseName );
+
+	//! Enabled/disables export of loggers
+	void SetDebuggingEnabled( bool bEnabled );
+
+	//! Logging export flag getter
+	bool GetIsDebuggingEnabled() const;
 
 	//! Returns (static) size of ring buffer
 	/**
@@ -167,6 +188,7 @@ public:
 	  */
 	void IncrementBlockPointer();
 
+
 protected:
 	//! This method is called by the networkg client and pushes samples into the ring buffer
 	/**
@@ -180,7 +202,7 @@ protected:
 	  *
 	  * @note This method is not called out of the audio streaming context but out of the network context.
 	  */
-	int Transmit( const ITASampleFrame& sfNewSamples, int iNumSamples );
+	int Transmit( const ITASampleFrame& sfNewSamples, const int iNumSamples );
 
 	//! Returns samples that can be read from ring buffer
 	/**
@@ -194,6 +216,9 @@ protected:
 	*/
 	int GetRingBufferFreeSamples() const;
 
+	//! Returns a string for the streaming status identifier
+	static std::string GetStreamingStatusString( const int iStreamingStatus );
+
 private:
 	CITANetAudioStreamingClient* m_pNetAudioStreamingClient; //!< Audio streaming network client
 
@@ -204,16 +229,17 @@ private:
 	int m_iWriteCursor; //!< Cursor where samples will be fed into ring buffer from net audio producer (always ahead)
 	bool m_bRingBufferFull; //!< Indicator if ring buffer is full (and read cursor equals write cursor)
 	ITASampleFrame m_sfRingBuffer; //!< Ring buffer
-	int m_iTargetSampleLatency; //!< Maximum allowed samples / target sample latency
 
 	int m_iStreamingStatus; //!< Current streaming status
 	double m_dLastStreamingTimeCode;
 
-	ITABufferedDataLoggerImplAudio* m_pAudioLogger; //!< Logging for the audio stream
-	ITABufferedDataLoggerImplStream* m_pStreamLogger; //!< Logging for the audio stream
-	ITABufferedDataLoggerImplNet* m_pNetLogger; //!< Logging for the network stream
-	int iAudioStreamingBlockID; //!< Audio streaming block id
-	int iNetStreamingBlockID; //!< Network streaming block id
+	ITABufferedDataLoggerImplStream* m_pAudioStreamLogger; //!< Logging for the audio stream
+	ITABufferedDataLoggerImplNet* m_pNetworkStreamLogger; //!< Logging for the network stream
+	std::string m_sNetAudioStreamLoggerBaseName;
+	bool m_bDebuggingEnabled;
+
+	int m_iAudioStreamingBlockID; //!< Audio streaming block id
+	int m_iNetStreamingBlockID; //!< Network streaming block id
 	
 	friend class CITANetAudioStreamingClient;
 };

include/ITANetAudioStreamingServer.h

@@ -21,60 +21,72 @@
 
 #include <ITADataSourcesDefinitions.h>
 
-#include <ITANetAudioProtocol.h>
+#include <ITASampleFrame.h>
+#include <ITAStopWatch.h>
+
+#include <VistaInterProcComm/Concurrency/VistaThreadLoop.h>
 
 #include <string>
 #include <vector>
-#include <iostream>
-#include <fstream>
-
-#include <ITANetAudioProtocol.h>
-
-#include <VistaInterProcComm/Concurrency/VistaThreadLoop.h>
-#include <ITASampleFrame.h>
+#include <iostream>
 
 class ITADatasource;
 class CITANetAudioMessage;
 class CITANetAudioProtocol;
 class CITANetAudioServer;
 class CITANetAudioMessage;
-class VistaTCPSocket;
+class CITABufferedDataLoggerImplServer;
+class VistaConnectionIP;
 
-//! Network audio sample server (for connecting a net audio stream)
+
+//! Network audio streaming server (for connecting a net audio stream) with an ITADataSource connection
 /**
- * Audio sample transmitter for a networked signal source that can connect via TCP/IP.
- *
- * \sa CITANetAudioStream
- * \note not thread-safe
- */
+  * Audio sample transmitter for a networked signal source that can connect via TCP/IP.
+  *
+  * @sa CITANetAudioStream
+  * @note not thread-safe
+  */
 class ITA_DATA_SOURCES_API CITANetAudioStreamingServer : public VistaThreadLoop
 {
 public:
 
-	enum UpdateStrategy
-	{
-		AUTO = 1, //!< Automatic update rate based on sample rate and block length of client (default)
-		ADAPTIVE, //!< Adaptive update rate, adjusts for drifting clocks
-		CONSTANT, //!< Set a user-defined update rate (may cause forced pausing of sample feeding or dropouts on client side)
-	};
-
 	CITANetAudioStreamingServer();
-	virtual ~CITANetAudioStreamingServer() {};
+	~CITANetAudioStreamingServer();
 
-	bool Start( const std::string& sAddress, int iPort );
+	//! Start to listen on a socket (blocking)
+	bool Start( const std::string& sAddress, const int iPort, const double dTimeIntervalCientSendStatus, const bool bUseUDP = false );
 	bool IsClientConnected() const;
 	std::string GetNetworkAddress() const;
 	int GetNetworkPort() const;
-	bool LoopBody();
+
 	void Stop();
 
 	void SetInputStream( ITADatasource* pInStream );
-	
+
 	int GetNetStreamBlocklength() const;
-	int GetNetStreamNumberOfChannels() const;
-	double GetNetStreamSampleRate() const;
+	int GetNetStreamNumberOfChannels( ) const;
+	double GetNetStreamSampleRate( ) const;
+
+	double GetEstimatedCorrFactor( ) const;
+	void SetEstimatedCorrFactor( double dcorrFactor );
+
+
+	//! Enabled/disables export of loggers
+	void SetDebuggingEnabled( bool bEnabled );
+
+	//! Logging export flag getter
+	bool GetIsDebuggingEnabled() const;
+
+	int GetSendingBlockLength() const;
+	void SetSendingBlockLength( const int iSendingBlockLength );
 
-	void SetAutomaticUpdateRate();
+	void SetTargetLatencySamples( const int iTargetLatency );
+	int GetTargetLatencySamples() const;
+
+	void SetServerLogBaseName( const std::string& sBaseName );
+	std::string GetServerLogBaseName() const;
+
+	bool LoopBody();
 
 protected:
 	ITADatasource* GetInputStream() const;
@@ -85,12 +97,25 @@ private:
 	ITADatasource* m_pInputStream;
 	VistaConnectionIP* m_pConnection;
 
-	CITANetAudioProtocol::StreamingParameters m_oServerParams;
-	CITANetAudioMessage* m_pIncomingMessage;
-	CITANetAudioMessage* m_pOutgoingMessage;
+	CITANetAudioMessage* m_pMessage;
+
+	CITABufferedDataLoggerImplServer* m_pServerLogger;
+	std::string m_sServerLogBaseName;
+	ITAStopWatch m_swTryReadBlockStats, m_swTryReadAccessStats;
+	bool m_bDebuggingEnabled;
+
+	int m_iServerBlockId;
+	double m_dLastTimeStamp;
+	double m_dEstimatedCorrFactor;
+
+	int m_iTargetLatencySamples;
+	int m_iEstimatedClientRingBufferFreeSamples;
+	int m_iClientRingBufferSize;
+	int m_iSendingBlockLength;
+	int m_iMaxSendBlocks;
 
-	int m_iUpdateStrategy;
-	int m_iClientRingBufferFreeSamples;
+	double m_dStreamTimeStart; //!< Stream time start
+	long unsigned int m_nStreamSampleCounts; //!< Samples that has been streamed
 
 	friend class CITANetAudioServer;
 };

include/ITAStreamInfo.h

-// $Id: $
+/*
+ * ----------------------------------------------------------------
+ *
+ *		ITA core libs
+ *		(c) Copyright Institute of Technical Acoustics (ITA)
+ *		RWTH Aachen University, Germany, 2015-2017
+ *
+ * ----------------------------------------------------------------
+ *				    ____  __________  _______
+ *				   //  / //__   ___/ //  _   |
+ *				  //  /    //  /    //  /_|  |
+ *				 //  /    //  /    //  ___   |
+ *				//__/    //__/    //__/   |__|
+ *
+ * ----------------------------------------------------------------
+ *
+ */
 
-#ifndef __ITA_STREAM_INFO_H__
-#define __ITA_STREAM_INFO_H__
+#ifndef INCLUDE_WATCHER_ITA_STREAM_INFO
+#define INCLUDE_WATCHER_ITA_STREAM_INFO
 
 #include <ITATypes.h>
 
-// Diese Datenklasse beschreibt den Zustand eines Audiostreams
-class ITAStreamInfo {
+//! Time code information for audio streams
+class ITAStreamInfo
+{
 public:
-	// Anzahl der abgespielten Samples seit Beginn des Streamings
-	uint64_t nSamples;
+	uint64_t nSamples; //!< Number of samples processed
+	double dStreamTimeCode; //!< Stream time code (starts with zero)
+	double dSysTimeCode; //!< System time stamp code (begings with current time stamp of system)
 
-	// TODO: Beschreiben
-	double dTimecode;
+	inline ITAStreamInfo()
+		: nSamples( 0 )
+		, dStreamTimeCode( 0.0f )
+		, dSysTimeCode( 0.0f )
+	{};
 
-	//! Standard-Konstruktor (setzt alle Werte 0)
-	ITAStreamInfo() : nSamples(0), dTimecode(0) {}
-
-    //! Destruktor
-	virtual ~ITAStreamInfo() {};
+	inline virtual ~ITAStreamInfo() {};
 };
 
-#endif // __ITA_STREAM_INFO_H__
+#endif // INCLUDE_WATCHER_ITA_STREAM_INFO

include/ITAStreamPatchBay.h

@@ -220,7 +220,7 @@ private:
 		ITADatasource* pDatasource;				//!< Datasource assigned to the input
 		std::vector< const float* > vpfInputData;	//!< Pointers to the next stream blocks
 
-		inline InputDesc( const int iChannels, const int iBlockLength )
+		inline InputDesc( const int iChannels, const int )
 			: vpfInputData( iChannels, nullptr )
 			, iChannels( iChannels )
 			, fCurrentGain( 1.0f )

src/ITABufferDataSink.cpp

@@ -6,6 +6,7 @@
 #include <ITAException.h>
 #include <ITAFunctors.h>
 #include <ITANumericUtils.h>
+#include <ITAClock.h>
 
 #ifndef WIN32
 #include <memory.h>
@@ -98,7 +99,8 @@ void ITABufferDataSink::Transfer( unsigned int uiSamples )
 
 			m_uiWriteCursor += m;
 			m_siState.nSamples += m_pdsSource->GetBlocklength();
-			m_siState.dTimecode = ( double ) ( m_siState.nSamples ) / m_pdsSource->GetSampleRate();
+			m_siState.dStreamTimeCode = ( double ) ( m_siState.nSamples ) / m_pdsSource->GetSampleRate();
+			m_siState.dSysTimeCode = ITAClock::getDefaultClock()->getTime();
 
 			m_pdsSource->IncrementBlockPointer();
 		}

src/ITADataSourceRealization.cpp

-
 #include "ITADataSourceRealization.h"
-
-#include <cassert>
 #include <ITAFastMath.h>
+#include <cassert>
 
-/*
-ITADatasourceRealization::ITADatasourceRealization(unsigned int uiChannels,
-		                                           unsigned int uiBlocklength,
-				                                   unsigned int uiCapacity)
-{
-	Init(uiChannels, uiBlocklength, uiCapacity);
-}
-*/
-
-ITADatasourceRealization::ITADatasourceRealization(unsigned int uiChannels,
-						                           double dSamplerate,
-		                                           unsigned int uiBlocklength,
-												   unsigned int uiCapacity)
+ITADatasourceRealization::ITADatasourceRealization( unsigned int uiChannels, double dSamplerate, unsigned int uiBlocklength, unsigned int uiCapacity )
 {
 	assert( dSamplerate > 0 );
 	m_dSampleRate = dSamplerate;
 	m_oStreamProps.dSamplerate = dSamplerate;
 
-	Init(uiChannels, uiBlocklength, uiCapacity);
-}	
+	Init( uiChannels, uiBlocklength, uiCapacity );
+}
 
-void ITADatasourceRealization::Init(unsigned int uiChannels,
-			                        unsigned int uiBlocklength,
-			                        unsigned int uiCapacity)
+void ITADatasourceRealization::Init( unsigned int uiChannels, unsigned int uiBlocklength, unsigned int uiCapacity )
 {
 	assert( uiChannels > 0 );
 	assert( uiBlocklength > 0 );
@@ -49,7 +33,7 @@ void ITADatasourceRealization::Init(unsigned int uiChannels,
 	m_oStreamProps.uiChannels = m_uiChannels;
 	m_oStreamProps.uiBlocklength = m_uiBlocklength;
 
-	m_uiBufferSize = uiBlocklength * (uiCapacity+1);
+	m_uiBufferSize = uiBlocklength * ( uiCapacity + 1 );
 
 	m_pEventHandler = NULL;
 
@@ -57,31 +41,33 @@ void ITADatasourceRealization::Init(unsigned int uiChannels,
 	   Organisation des Puffers: Damit die Blcke der einzelnen Kanle
 	   im Speicher ortlich nher liegen ist das Array wiefolgt indiziert:
 
-       [1. Block Kanal 1], ..., [1. Block Kanal k], [2. Block Kanal 1], ...
+	   [1. Block Kanal 1], ..., [1. Block Kanal k], [2. Block Kanal 1], ...
 
-	*/
+	   */
 
 	// Puffer erzeugen und mit Nullen initialiseren
 	// TODO: Fehlerbehandlung beim Speicherallozieren
 	/* Bugfix zu Bug #001:
-	   
+
 	   Hier wurde der Puffer einfach um 1024 Felder verlngert.
 	   Damit Funktioniert Wuschels ASIO4ALL jetzt. Ungeklrt aber
 	   warum der Fehler auftrat?
 
-       2005-2-14
-	*/
+	   2005-2-14
+	   */
 
-	m_pfBuffer = fm_falloc(m_uiBufferSize * m_uiChannels + /* >>> */ 1024 /* <<< */, false);		
+	m_pfBuffer = fm_falloc( m_uiBufferSize * m_uiChannels + /* >>> */ 1024 /* <<< */, false );
 
-	Reset();	
+	Reset();
 }
 
-ITADatasourceRealization::~ITADatasourceRealization() {
-	fm_free(m_pfBuffer);
+ITADatasourceRealization::~ITADatasourceRealization()
+{
+	fm_free( m_pfBuffer );
 }
 
-void ITADatasourceRealization::Reset() {
+void ITADatasourceRealization::Reset()
+{
 	m_uiReadCursor = 0;
 	m_uiWriteCursor = 0;
 
@@ -93,22 +79,26 @@ void ITADatasourceRealization::Reset() {
 	m_iGBPEntrances = 0;
 	m_bGBPFirst = true;
 
-	fm_zero(m_pfBuffer, m_uiBufferSize * m_uiChannels + /* >>> */ 1024 /* <<< */);
+	fm_zero( m_pfBuffer, m_uiBufferSize * m_uiChannels + /* >>> */ 1024 /* <<< */ );
 }
 
-bool ITADatasourceRealization::HasStreamErrors() const {
-	return (m_iBufferUnderflows > 0) || (m_iBufferOverflows > 0) || (m_iGBPReentrances > 0);
+bool ITADatasourceRealization::HasStreamErrors() const
+{
+	return ( m_iBufferUnderflows > 0 ) || ( m_iBufferOverflows > 0 ) || ( m_iGBPReentrances > 0 );
 }
 
-ITADatasourceRealizationEventHandler* ITADatasourceRealization::GetStreamEventHandler() const {
+ITADatasourceRealizationEventHandler* ITADatasourceRealization::GetStreamEventHandler() const
+{
 	return m_pEventHandler;
 }
 
-void ITADatasourceRealization::SetStreamEventHandler(ITADatasourceRealizationEventHandler* pHandler) {
+void ITADatasourceRealization::SetStreamEventHandler( ITADatasourceRealizationEventHandler* pHandler )
+{
 	m_pEventHandler = pHandler;
 }
 
-const float* ITADatasourceRealization::GetBlockPointer(unsigned int uiChannel, const ITAStreamInfo* pStreamInfo) {
+const float* ITADatasourceRealization::GetBlockPointer( unsigned int uiChannel, const ITAStreamInfo* pStreamInfo )
+{
 	assert( uiChannel < m_uiChannels );
 
 	/*
@@ -117,7 +107,8 @@ const float* ITADatasourceRealization::GetBlockPointer(unsigned int uiChannel, c
 	 *
 	 * WICHTIG: Dies sollte nicht passieren. Fehler beim anwendenden Programmierer!
 	 */
-	if (++m_iGBPEntrances > 1) {
+	if( ++m_iGBPEntrances > 1 )
+	{
 		--m_iGBPEntrances;
 		++m_iGBPReentrances;
 		return NULL;
@@ -125,12 +116,16 @@ const float* ITADatasourceRealization::GetBlockPointer(unsigned int uiChannel, c
 
 	// Hook/Handler aufrufen
 	PreGetBlockPointer();
-	if (m_pEventHandler) m_pEventHandler->HandlePreGetBlockPointer(this, uiChannel);
+	if( m_pEventHandler )
+		m_pEventHandler->HandlePreGetBlockPointer( this, uiChannel );
 
-	if (m_bGBPFirst) {
+	if( m_bGBPFirst )
+	{
 		// Erster Eintritt in GBP seit letztem IBP => Daten produzieren
-		ProcessStream(pStreamInfo);
-		if (m_pEventHandler) m_pEventHandler->HandleProcessStream(this, pStreamInfo);
+		ProcessStream( pStreamInfo );
+
+		if( m_pEventHandler )
+			m_pEventHandler->HandleProcessStream( this, pStreamInfo );
 
 		m_bGBPFirst = false;
 	}
@@ -145,45 +140,51 @@ const float* ITADatasourceRealization::GetBlockPointer(unsigned int uiChannel, c
 	 */
 
 	unsigned int uiLocalReadCursor = m_uiReadCursor;
-	if (uiLocalReadCursor == m_uiWriteCursor) {
+	if( uiLocalReadCursor == m_uiWriteCursor )
+	{
 		++m_iBufferUnderflows;
 		--m_iGBPEntrances;
 		return NULL;
 	}
 
 	--m_iGBPEntrances;
-	return m_pfBuffer + (uiChannel * m_uiBufferSize) + uiLocalReadCursor;
+	return m_pfBuffer + ( uiChannel * m_uiBufferSize ) + uiLocalReadCursor;
 }
 
-void ITADatasourceRealization::IncrementBlockPointer() {
-
+void ITADatasourceRealization::IncrementBlockPointer()
+{
 	unsigned int uiLocalReadCursor = m_uiReadCursor;
 
-	if (uiLocalReadCursor == m_uiWriteCursor)
+	if( uiLocalReadCursor == m_uiWriteCursor )
 		// Keine Daten im Ausgabepuffer? Kein Inkrement mglich! (Fehlerfall)
 		++m_iBufferUnderflows;
 	else
 		// Lesezeiger inkrementieren
-		m_uiReadCursor = (uiLocalReadCursor + m_uiBlocklength) % m_uiBufferSize;
+		m_uiReadCursor = ( uiLocalReadCursor + m_uiBlocklength ) % m_uiBufferSize;
 
 	m_bGBPFirst = true;
 
 	PostIncrementBlockPointer();
-	if (m_pEventHandler) m_pEventHandler->HandlePostIncrementBlockPointer(this);
+
+	if( m_pEventHandler )
+		m_pEventHandler->HandlePostIncrementBlockPointer( this );
 }
 
-float* ITADatasourceRealization::GetWritePointer(unsigned int uiChannel) {
+float* ITADatasourceRealization::GetWritePointer( unsigned int uiChannel )
+{
 	assert( uiChannel < m_uiChannels );
-	return m_pfBuffer + (uiChannel * m_uiBufferSize) + m_uiWriteCursor;
+	return m_pfBuffer + ( uiChannel * m_uiBufferSize ) + m_uiWriteCursor;
 }
 
-void ITADatasourceRealization::IncrementWritePointer() {
+void ITADatasourceRealization::IncrementWritePointer()
+{
 	// Lokaler Schreibcursor
 	unsigned int uiLocalWriteCursor = m_uiWriteCursor;
-	unsigned int uiNewWriteCursor = (uiLocalWriteCursor + m_uiBlocklength) % m_uiBufferSize;
+	unsigned int uiNewWriteCursor = ( uiLocalWriteCursor + m_uiBlocklength ) % m_uiBufferSize;
 
 	// Pufferberlauf
-	if (uiNewWriteCursor == m_uiReadCursor) {
+	if( uiNewWriteCursor == m_uiReadCursor )
+	{
 		++m_iBufferOverflows;
 		return;
 	}
@@ -192,6 +193,6 @@ void ITADatasourceRealization::IncrementWritePointer() {
 	m_uiWriteCursor = uiNewWriteCursor;
 }
 
-void ITADatasourceRealizationEventHandler::HandlePreGetBlockPointer(ITADatasourceRealization* pSender, unsigned int uiChannel) {}
-void ITADatasourceRealizationEventHandler::HandlePostIncrementBlockPointer(ITADatasourceRealization* pSender) {}
-void ITADatasourceRealizationEventHandler::HandleProcessStream(ITADatasourceRealization* pSender, const ITAStreamInfo* pStreamInfo) {}
+void ITADatasourceRealizationEventHandler::HandlePreGetBlockPointer( ITADatasourceRealization*, unsigned int ) {}
+void ITADatasourceRealizationEventHandler::HandlePostIncrementBlockPointer( ITADatasourceRealization* ) {}
+void ITADatasourceRealizationEventHandler::HandleProcessStream( ITADatasourceRealization*, const ITAStreamInfo* ) {}

src/ITADataSourceUtils.cpp

@@ -8,6 +8,8 @@
 #include <ITAStreamInfo.h>
 #include <ITAAudiofileWriter.h>
 #include <ITAException.h>
+#include <ITAClock.h>
+
 #include <cmath>
 #include <string>
 #include <vector>
@@ -90,7 +92,8 @@ void WriteFromDatasourceToBuffer(ITADatasource* pSource,
 			n += uiBlocklength;
 
 			siState.nSamples += uiBlocklength;
-			siState.dTimecode = (double) (siState.nSamples) / dSamplerate;
+			siState.dStreamTimeCode = (double) (siState.nSamples) / dSamplerate;
+			siState.dSysTimeCode = ITAClock::getDefaultClock()->getTime();
 
 			if (bDisplayProgress) 
 			{
@@ -193,7 +196,8 @@ void WriteFromDatasourceToFile(ITADatasource* pSource,
 			pSource->IncrementBlockPointer();
 	
 			siState.nSamples += uiBlocklength;
-			siState.dTimecode = (double) (siState.nSamples) / dSamplerate;
+			siState.dStreamTimeCode = (double) (siState.nSamples) / dSamplerate;
+			siState.dSysTimeCode = ITAClock::getDefaultClock()->getTime();
 
 			// Daten schreiben
 			writer->write((std::min)(uiBlocklength, (uiNumberOfSamples - n)), vpfData);

src/ITAFileDataSink.cpp

@@ -4,13 +4,16 @@
 #include <ITADataSource.h>
 #include <ITAAudiofileWriter.h>
 #include <ITANumericUtils.h>
+#include <ITAClock.h>
 
 ITAFileDatasink::ITAFileDatasink( std::string sFilename, ITADatasource* pdsSource, ITAQuantization eQuantization )
-	: m_pfSilence( NULL ) {
+	: m_pfSilence( NULL )
+{
 	m_pdsSource = pdsSource;
 	m_pFileWriter = NULL;
 
-	if( pdsSource ) {
+	if( pdsSource )
+	{
 		m_vpfData.resize( pdsSource->GetNumberOfChannels() );
 
 		ITAAudiofileProperties props;
@@ -24,19 +27,23 @@ ITAFileDatasink::ITAFileDatasink( std::string sFilename, ITADatasource* pdsSourc
 	}
 }
 
-ITAFileDatasink::~ITAFileDatasink() {
+ITAFileDatasink::~ITAFileDatasink()
+{
 	delete m_pFileWriter;
 	fm_free( m_pfSilence );
 }
 
-void ITAFileDatasink::Transfer( unsigned int uiSamples ) {
-	if( m_pdsSource ) {
+void ITAFileDatasink::Transfer( unsigned int uiSamples )
+{
+	if( m_pdsSource )
+	{
 		// Anzahl der zu transferrierenden Blcke bestimmen
 		unsigned int b = m_pdsSource->GetBlocklength();
 		unsigned int n = uprdivu( uiSamples, b );
 
 		for( unsigned int i = 0; i < n; i++ ) {
-			for( unsigned int j = 0; j < m_pdsSource->GetNumberOfChannels(); j++ ) {
+			for( unsigned int j = 0; j < m_pdsSource->GetNumberOfChannels(); j++ )
+			{
 				const float* pfSrc = m_pdsSource->GetBlockPointer( j, &m_siState );
 				if( pfSrc )
 					m_vpfData[ j ] = ( float* ) pfSrc;
@@ -45,7 +52,8 @@ void ITAFileDatasink::Transfer( unsigned int uiSamples ) {
 			}
 			m_pdsSource->IncrementBlockPointer();
 			m_siState.nSamples += b;
-			m_siState.dTimecode = ( double ) ( m_siState.nSamples ) / m_pdsSource->GetSampleRate();
+			m_siState.dStreamTimeCode = ( double ) ( m_siState.nSamples ) / m_pdsSource->GetSampleRate();
+			m_siState.dSysTimeCode = ITAClock::getDefaultClock()->getTime();
 
 			m_pFileWriter->write( b, m_vpfData );
 		}

src/ITANetAudioClient.cpp

-#include <ITANetAudioClient.h>
+#include "ITANetAudioClient.h"
 
-#include <ITANetAudioMessage.h>
-#include <ITANetAudioProtocol.h>
 #include <ITANetAudioStream.h>
+#include <ITAException.h>
 
 #include <VistaInterProcComm/Connections/VistaConnectionIP.h>
 
@@ -16,13 +15,15 @@ CITANetAudioClient::~CITANetAudioClient()
 	delete m_pConnection;
 }
 
-bool CITANetAudioClient::Connect( const std::string& sAddress, int iPort )
+bool CITANetAudioClient::Connect( const std::string& sAddress, const int iPort, const bool bUseUDP /* = false */ )
 {
 	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 );
+	const VistaConnectionIP::VistaProtocol iCTProtocol = bUseUDP ? VistaConnectionIP::CT_UDP : VistaConnectionIP::CT_TCP;
+	m_pConnection = new VistaConnectionIP( iCTProtocol, sAddress, iPort );
+	
 	if( !GetIsConnected() )
 	{
 		delete m_pConnection;
@@ -46,5 +47,8 @@ void CITANetAudioClient::Disconnect()
 
 bool CITANetAudioClient::GetIsConnected() const
 {
-	return m_pConnection ? true : false;
+	if( m_pConnection )
+		return m_pConnection->GetIsOpen();
+	else
+		return false;
 }

include/ITANetAudioClient.h → src/ITANetAudioClient.h

@@ -28,7 +28,7 @@ class VistaConnectionIP;
 //! A network audio client that connects to a network audio server
 /** 
   * Use CITANetAudioStreamingClient to start an audio stream with the connection of this client.
-  * This class is basically a helper around Vista TCP/IP network functionality.
+  * This class is basically a helper around Vista TCP/IP or UDP network functionality.
   *
   */
 class CITANetAudioClient
@@ -42,7 +42,7 @@ public:
 	CITANetAudioClient();
 	~CITANetAudioClient();
 
-	bool Connect( const std::string& sAddress, int iPort );
+	bool Connect( const std::string& sAddress, const int iPort, const bool bUseUDP );
 	void Disconnect();
 	bool GetIsConnected() const;
 

src/ITANetAudioMessage.cpp

-#include <ITANetAudioMessage.h>
+#include "ITANetAudioMessage.h"
+
+#include <ITAClock.h>
+#include <ITADataLog.h>
 #include <ITAStringUtils.h>
 
 #include <VistaInterProcComm/Connections/VistaConnectionIP.h>
 #include <VistaBase/VistaExceptionBase.h>
 #include <VistaBase/VistaStreamUtils.h>
 
-#include <cstring>
 #include <algorithm>
+#include <cstring>
 #include <cassert>
 #include <iostream>
 #include <iomanip>
 
 static int S_nMessageIds = 0;
 
-CITANetAudioMessage::CITANetAudioMessage( VistaConnectionIP* pConnection )
+struct ITANetAudioMessageLog : public ITALogDataBase
+{
+	inline static std::ostream& outputDesc( std::ostream& os )
+	{
+		os << "BlockId";
+		os << "\t" << "WorldTimeStamp";
+		os << "\t" << "MessageType";
+		os << "\t" << "Action";
+		os << "\t" << "InternalProcessingTime";
+		os << "\t" << "PayloadSize";
+		os << std::endl;
+		return os;
+	};
+
+	inline std::ostream& outputData( std::ostream& os ) const
+	{
+		os << uiBlockId;
+		os << "\t" << std::setprecision( 12 ) << dWorldTimeStamp;
+		os << "\t" << sMessageType;
+		os << "\t" << sAction;
+		os << "\t" << std::setprecision( 12 ) << dInternalProcessingTime;
+		os << "\t" << nMessagePayloadSize;
+		os << std::endl;
+		return os;
+	};
+
+	unsigned int uiBlockId; //!< Block identifier (audio streaming)
+	double dWorldTimeStamp; //!< Time stamp at beginning of logged message process
+	std::string sMessageType; //!< Protocol message type
+	std::string sAction; //!< Triggered action
+	double dInternalProcessingTime; //!< Processing within message class
+	VistaType::sint32 nMessagePayloadSize; //!< Data
+
+};
+
+class ITABufferedDataLoggerImplProtocol : public ITABufferedDataLogger < ITANetAudioMessageLog > {};
+
+CITANetAudioMessage::CITANetAudioMessage( VistaSerializingToolset::ByteOrderSwapBehavior bSwapBuffers )
 	: m_vecIncomingBuffer( 2048 )
 	, m_oOutgoing( 2048 )
-	, m_pConnection( pConnection )
-	, m_nTimeoutMilliseconds( 1 )
+	, m_pConnection( NULL )
+	, m_iBytesReceivedTotal( 0 )
+	, m_sMessageLoggerBaseName( "ITANetAudioMessage" )
+	, m_bDebuggingEnabled( false )
 {
-	m_oOutgoing.SetByteorderSwapFlag( pConnection->GetByteorderSwapFlag() );
-	m_oIncoming.SetByteorderSwapFlag( pConnection->GetByteorderSwapFlag() );
+	m_pMessageLogger = new ITABufferedDataLoggerImplProtocol();
+	m_pMessageLogger->setOutputFile( m_sMessageLoggerBaseName + ".log" );
+
+	m_nMessageId = 0;
+	m_oOutgoing.SetByteorderSwapFlag( bSwapBuffers );
+	m_oIncoming.SetByteorderSwapFlag( bSwapBuffers );
+
 	ResetMessage();
 }
 
 void CITANetAudioMessage::ResetMessage()
 {
+	const double dInTime = ITAClock::getDefaultClock()->getTime();
+
+	ITANetAudioMessageLog oLog;
+	oLog.uiBlockId = m_nMessageId;
+	oLog.sMessageType = "RESET_MESSAGE";
+	oLog.nMessagePayloadSize = 0;
+	oLog.dWorldTimeStamp = dInTime;
+
+	oLog.sAction = "reset_message";
+
 	if( m_oIncoming.GetTailSize() > 0 )
+	{
 		vstr::err() << "CITANetAudioMessage::ResetMessage() called before message was fully processed!" << std::endl;
+		oLog.sAction = "reset_failed";
+	}
 
 	// wait till sending is complete -> this prevents us
 	// from deleting the buffer while it is still being read
 	// by the connection
-	if( m_pConnection )
-		m_pConnection->WaitForSendFinish();
+	if( m_pConnection != NULL )
+		m_pConnection->WaitForSendFinish(); // can be time-costly
 
 	m_nMessageId = S_nMessageIds++;
 
@@ -44,20 +104,32 @@ void CITANetAudioMessage::ResetMessage()
 
 	m_oIncoming.SetBuffer( NULL, 0 );
 
-	m_nMessageType = CITANetAudioProtocol::NP_INVALID;
+	m_nMessageType = -1;
+
+	oLog.dInternalProcessingTime = ITAClock::getDefaultClock()->getTime() - dInTime;
+	m_pMessageLogger->log( oLog );
 
 #if NET_AUDIO_SHOW_TRAFFIC
 	vstr::out() << "CITANetAudioMessage [Preparing] (id=" << std::setw( 4 ) << m_nMessageId << ")" << std::endl;
 #endif
 }
 
+void CITANetAudioMessage::SetConnection( VistaConnectionIP* pConn )
+{
+	m_pConnection = pConn;
+}
+
 void CITANetAudioMessage::WriteMessage()
 {
+	const double dInTime = ITAClock::getDefaultClock()->getTime();
+	ITANetAudioMessageLog oLog;
+	oLog.dWorldTimeStamp = dInTime;
 	VistaType::byte* pBuffer = ( VistaType::byte* ) m_oOutgoing.GetBuffer();
 	VistaType::sint32 iSwapDummy;
 
 	// rewrite size dummy
 	iSwapDummy = m_oOutgoing.GetBufferSize() - sizeof( VistaType::sint32 );
+	oLog.nMessagePayloadSize = iSwapDummy;
 	if( m_oOutgoing.GetByteorderSwapFlag() )
 		VistaSerializingToolset::Swap4( &iSwapDummy );
 	std::memcpy( pBuffer, &iSwapDummy, sizeof( VistaType::sint32 ) );
@@ -66,6 +138,7 @@ void CITANetAudioMessage::WriteMessage()
 
 	// rewrite type dummy
 	iSwapDummy = m_nMessageType;
+	oLog.sMessageType = CITANetAudioProtocol::GetNPMessageID( m_nMessageType );
 	if( m_oOutgoing.GetByteorderSwapFlag() )
 		VistaSerializingToolset::Swap4( &iSwapDummy );
 	std::memcpy( pBuffer, &iSwapDummy, sizeof( VistaType::sint32 ) );
@@ -74,9 +147,11 @@ void CITANetAudioMessage::WriteMessage()
 
 	// rewrite messageid dummy
 	iSwapDummy = m_nMessageId;
+	oLog.uiBlockId = m_nMessageId;
 	if( m_oOutgoing.GetByteorderSwapFlag() )
 		VistaSerializingToolset::Swap4( &iSwapDummy );
 	std::memcpy( pBuffer, &iSwapDummy, sizeof( VistaType::sint32 ) );
+	oLog.sAction = "write_message";
 
 #if NET_AUDIO_SHOW_TRAFFIC
 	vstr::out() << "CITANetAudioMessage [  Writing] " << m_nMessageType << " (id=" << std::setw( 4 ) << m_nMessageId << ")" << std::endl;
@@ -86,75 +161,106 @@ void CITANetAudioMessage::WriteMessage()
 	{
 		// It appears safe to send even very big data payload, so we will send at once
 		int iRawBufferSize = m_oOutgoing.GetBufferSize();
-		assert( iRawBufferSize > 4 );
 		int nRet = m_pConnection->Send( m_oOutgoing.GetBuffer(), iRawBufferSize );
 
 #if NET_AUDIO_SHOW_TRAFFIC
 		vstr::out() << "CITANetAudioMessage [  Writing] " << m_nMessageType << " (id=" << std::setw( 4 ) << m_nMessageId << ") RAW BUFFER DONE" << std::endl;
 #endif
 
-		m_pConnection->WaitForSendFinish();
+		// Block processing until data is successfully transmitted
+		unsigned long nData = m_pConnection->WaitForSendFinish( 0 );
 		if( nRet != m_oOutgoing.GetBufferSize() )
 			VISTA_THROW( "ITANetAudioMessage: could not send all data from output buffer via network connection", 255 );
 	}
-	catch( VistaExceptionBase& ex )
+	catch (VistaExceptionBase& ex)
 	{
 		ITA_EXCEPT1( NETWORK_ERROR, ex.GetExceptionText() );
 	}
+
+	oLog.dInternalProcessingTime = ITAClock::getDefaultClock()->getTime() - dInTime;
+	m_pMessageLogger->log( oLog );
+
+	return;
 }
 
-bool CITANetAudioMessage::TryReadMessage()
+bool CITANetAudioMessage::ReadMessage( const int iTimeoutMilliseconds )
 {
+	ITANetAudioMessageLog oLog;
+	const double dInTime = ITAClock::getDefaultClock()->getTime();
+	oLog.dWorldTimeStamp = dInTime;
+
 #if NET_AUDIO_SHOW_TRAFFIC
-	vstr::out() << "CITANetAudioMessage [ TryRead ] Waiting for incoming data for " << m_nTimeoutMilliseconds << std::endl;
-#endif
-	long nIncomingBytes = m_pConnection->WaitForIncomingData( m_nTimeoutMilliseconds );
-	if( nIncomingBytes <= 0 )
-	{
-#if NET_AUDIO_SHOW_TRAFFIC
-			vstr::out() << "CITANetAudioMessage [ TryRead ] nothing incoming" << std::endl;
+	vstr::out() << "CITANetAudioMessage [ Reading ] Waiting for incoming data" << std::endl;
 #endif
+
+	// WaitForIncomming Data int in ca ms
+	long nIncomingBytes = m_pConnection->WaitForIncomingData( iTimeoutMilliseconds );
+
+	if( nIncomingBytes == -1 )
 		return false;
-	}
-		
-	assert( nIncomingBytes >= 4 ); // we need at least the size of message
+
 #if NET_AUDIO_SHOW_TRAFFIC
 	vstr::out() << "CITANetAudioMessage [ Reading ] " << nIncomingBytes << " bytes incoming" << std::endl;
 #endif
-
+	
 	VistaType::sint32 nMessagePayloadSize;
 	int nBytesRead = m_pConnection->ReadInt32( nMessagePayloadSize );
 	assert( nBytesRead == sizeof( VistaType::sint32 ) );
+	oLog.nMessagePayloadSize = nMessagePayloadSize;
+
 #if NET_AUDIO_SHOW_TRAFFIC
 	vstr::out() << "CITANetAudioMessage [ Reading ] Expecting " << nMessagePayloadSize << " bytes message payload" << std::endl;
 #endif
-	// we need at least the two protocol ints
-	//assert( nMessagePayloadSize >= 2 * sizeof( VistaType::sint32 ) );
+	
+	if( nMessagePayloadSize <= 0 )
+		return false; // Try-read failed, returning!
 
+	// we need at least the two protocol ints
+	assert( nMessagePayloadSize >= 2 * sizeof( VistaType::sint32 ) );
+	
 	if( nMessagePayloadSize > ( int ) m_vecIncomingBuffer.size() )
 		m_vecIncomingBuffer.resize( nMessagePayloadSize );
 
-	// Receive all incoming data (potentially splitted)
-	int iBytesReceivedTotal = 0;
-	while( nMessagePayloadSize < iBytesReceivedTotal )
+
+	// Receive all incoming data (potentially splitted!!)
+
+	m_iBytesReceivedTotal = 0;
+	while( ( unsigned long ) nMessagePayloadSize != m_iBytesReceivedTotal )
 	{
+		// We force a blocking-wait for the rest (or the data to be fetched until message payload is copied)
 		int iIncommingBytes = m_pConnection->WaitForIncomingData( 0 );
-		int iBytesReceived = m_pConnection->Receive( &m_vecIncomingBuffer[ iBytesReceivedTotal ], iIncommingBytes );
-		iBytesReceivedTotal += iBytesReceived;
+		assert( m_iBytesReceivedTotal < m_vecIncomingBuffer.size() );
+
+		// Check if we are already receiving another message and only read until end-of-payload, then release this message-read
+		int iBytesReceived = -1;
+		const int nPendingPayloadBytes = nMessagePayloadSize - m_iBytesReceivedTotal;
+		if( iIncommingBytes > nPendingPayloadBytes )
+			iBytesReceived = m_pConnection->Receive( &m_vecIncomingBuffer[ m_iBytesReceivedTotal ], nPendingPayloadBytes );
+		else
+			iBytesReceived = m_pConnection->Receive( &m_vecIncomingBuffer[ m_iBytesReceivedTotal ], iIncommingBytes );
+
+		m_iBytesReceivedTotal += iBytesReceived;
+
 #if NET_AUDIO_SHOW_TRAFFIC
-		vstr::out() << "[ CITANetAudioMessage ] " << std::setw( 3 ) << std::floor( iBytesReceivedTotal / float( nMessagePayloadSize ) * 100.0f ) << "% transmitted" << std::endl;
+		vstr::out() << "[ CITANetAudioMessage ] " << std::setw( 3 ) << std::floor( m_iBytesReceivedTotal / float( nMessagePayloadSize ) * 100.0f ) << "% transmitted" << std::endl;
 #endif
 	}
+	assert( m_iBytesReceivedTotal == nMessagePayloadSize );
 
 	// Transfer data into members
 	m_oIncoming.SetBuffer( &m_vecIncomingBuffer[ 0 ], nMessagePayloadSize, false );
 	m_nMessageType = ReadInt();
 	m_nMessageId = ReadInt();
 
+	oLog.sAction = "read_message";
+	oLog.sMessageType = CITANetAudioProtocol::GetNPMessageID( m_nMessageType );
+	oLog.uiBlockId = m_nMessageId;
+	oLog.dWorldTimeStamp = ITAClock::getDefaultClock()->getTime() - dInTime;
+	m_pMessageLogger->log( oLog );
+
 #if NET_AUDIO_SHOW_TRAFFIC
 	vstr::out() << "CITANetAudioMessage [ Reading ] Finished receiving " << m_nMessageType << " (id=" << std::setw( 4 ) << m_nMessageId << ")" << std::endl;
 #endif
-
 	return true;
 }
 
@@ -165,7 +271,6 @@ int CITANetAudioMessage::GetMessageType() const
 
 void CITANetAudioMessage::SetMessageType( int nType )
 {
-	assert( m_nMessageType == CITANetAudioProtocol::NP_INVALID ); // should only be set once
 	m_nMessageType = nType;
 }
 
@@ -278,11 +383,15 @@ VistaConnectionIP* CITANetAudioMessage::GetConnection() const
 	return m_pConnection;
 }
 
-void CITANetAudioMessage::ClearConnection() {
+void CITANetAudioMessage::ClearConnection()
+{
 	m_pConnection = NULL;
+	if( GetIsDebuggingEnabled() == false )
+		m_pMessageLogger->setOutputFile( "" ); // disable output
+	delete m_pMessageLogger;
 }
 
-void CITANetAudioMessage::WriteIntVector( const std::vector<int> viData )
+void CITANetAudioMessage::WriteIntVector( const std::vector< int > viData )
 {
 	int iSize = ( int ) viData.size();
 	WriteInt( iSize );
@@ -307,6 +416,7 @@ CITANetAudioProtocol::StreamingParameters CITANetAudioMessage::ReadStreamingPara
 	oParams.iChannels = ReadInt();
 	oParams.dSampleRate = ReadDouble();
 	oParams.iBlockSize = ReadInt();
+	oParams.iRingBufferSize = ReadInt();
 
 	return oParams;
 }
@@ -316,6 +426,7 @@ void CITANetAudioMessage::WriteStreamingParameters( const CITANetAudioProtocol::
 	WriteInt( oParams.iChannels );
 	WriteDouble( oParams.dSampleRate );
 	WriteInt( oParams.iBlockSize );
+	WriteInt( oParams.iRingBufferSize );
 }
 
 int CITANetAudioMessage::ReadRingBufferSize()
@@ -342,17 +453,12 @@ void CITANetAudioMessage::ReadSampleFrame( ITASampleFrame* pSampleFrame )
 {
 	int iChannels = ReadInt();
 	int iSamples = ReadInt();
-
 	if( pSampleFrame->channels() != iChannels || pSampleFrame->GetLength() != iSamples )
 		pSampleFrame->init( iChannels, iSamples, false );
 
 	for( int i = 0; i < iChannels; i++ )
-	{
 		for( int j = 0; j < iSamples; j++ )
-		{
 			( *pSampleFrame )[ i ][ j ] = ReadFloat();
-		}
-	}
 }
 
 void CITANetAudioMessage::WriteSampleFrame( ITASampleFrame* pSamples )
@@ -361,10 +467,28 @@ void CITANetAudioMessage::WriteSampleFrame( ITASampleFrame* pSamples )
 	WriteInt( pSamples->GetLength() );
 
 	for( int i = 0; i < pSamples->channels(); i++ )
-	{
 		for( int j = 0; j < pSamples->GetLength(); j++ )
-		{
 			WriteFloat( ( *pSamples )[ i ][ j ] );
-		}
-	}
+}
+
+void CITANetAudioMessage::SetMessageLoggerBaseName( const std::string& sBaseName )
+{
+	assert( !sBaseName.empty() );
+	if( m_pMessageLogger )
+		m_pMessageLogger->setOutputFile( sBaseName + ".log" );
+}
+
+std::string CITANetAudioMessage::GetMessageLoggerBaseName() const
+{
+	return m_sMessageLoggerBaseName;
+}
+
+void CITANetAudioMessage::SetDebuggingEnabled( bool bEnabled )
+{
+	m_bDebuggingEnabled = bEnabled;
+}
+
+bool CITANetAudioMessage::GetIsDebuggingEnabled() const
+{
+	return m_bDebuggingEnabled;
 }

include/ITANetAudioMessage.h → src/ITANetAudioMessage.h

@@ -21,11 +21,12 @@
 
 #include <ITADataSourcesDefinitions.h>
 
+#include "ITANetAudioProtocol.h"
+
 // ITA includes
 #include <ITAException.h>
 #include <ITASampleBuffer.h>
 #include <ITASampleFrame.h>
-#include <ITANetAudioProtocol.h>
 
 // Vista includes
 #include <VistaInterProcComm/Connections/VistaByteBufferSerializer.h>
@@ -36,6 +37,7 @@
 #include <vector>
 
 class VistaConnectionIP;
+class ITABufferedDataLoggerImplProtocol;
 
 //! Network audio messages
 /**
@@ -48,13 +50,17 @@ class VistaConnectionIP;
 class ITA_DATA_SOURCES_API CITANetAudioMessage
 {
 public:
-	CITANetAudioMessage( VistaConnectionIP* );
+	CITANetAudioMessage( VistaSerializingToolset::ByteOrderSwapBehavior bSwapBuffers );
 
+	void SetConnection( VistaConnectionIP* );
 	VistaConnectionIP* GetConnection() const;
 	void ClearConnection();
 
+	//! Will always block processing until data is completely send
 	void WriteMessage();
-	bool TryReadMessage();
+
+	//! Returns false if no incomming data during timeout
+	bool ReadMessage( const int iTimeoutMilliseconds );
 
 	void ResetMessage();
 
@@ -65,6 +71,7 @@ public:
 	void SetMessageType( int nType );
 	int GetMessageType() const;
 
+
 	void WriteInt( const int );
 	void WriteBool( const bool );
 	void WriteDouble( const double );
@@ -91,15 +98,25 @@ public:
 	int ReadRingBufferFree();
 	void ReadSampleFrame(ITASampleFrame* pSampleFrame);
 
+	void SetMessageLoggerBaseName( const std::string& );
+	std::string GetMessageLoggerBaseName() const;
+	void SetDebuggingEnabled( bool bEnabled );
+	bool GetIsDebuggingEnabled() const;
+
 private:
 	int m_nMessageType;
 	int m_nMessageId;
+	unsigned long m_iBytesReceivedTotal;
+
 	VistaByteBufferSerializer m_oOutgoing; //!< Serialization buffer for messages
 	VistaByteBufferDeSerializer m_oIncoming; //!< Deserialization buffer for messages
 	std::vector< VistaType::byte > m_vecIncomingBuffer; // Net IO buffer
 
-	int m_nTimeoutMilliseconds; //!< Timeout for try-read message
 	VistaConnectionIP* m_pConnection;
+
+	ITABufferedDataLoggerImplProtocol* m_pMessageLogger;
+	std::string m_sMessageLoggerBaseName;
+	bool m_bDebuggingEnabled;
 };
 
 #endif // INCLUDE_WATCHER_ITA_NET_AUDIO_MESSAGE

src/ITANetAudioProtocol.cpp

-#include <ITANetAudioProtocol.h>
+#include "ITANetAudioProtocol.h"
 
 // @todo remove (all inline implemented)
\ No newline at end of file