[DEV] clip in small source file and add some message

audio_core/CMakeLists.txt

@@ -79,11 +79,15 @@ include_directories(
 add_executable(audio_core_node
   src/debug.cpp
   src/main.cpp
+  src/InterfaceInput.cpp
+  src/InterfaceOutput.cpp
+  src/InterfaceOutputManager.cpp
+  src/InterfaceOutputElement.cpp
 )
 
 ## Add cmake target dependencies of the executable/library
 ## as an example, message headers may need to be generated before nodes
-add_dependencies(${PROJECT_NAME}_node audio_core_generate_messages_cpp)
+add_dependencies(${PROJECT_NAME}_node audio_msg_generate_messages_cpp)
 
 ## Specify libraries to link a library or executable target against
 target_link_libraries(${PROJECT_NAME}_node

audio_core/src/InterfaceInput.cpp

@@ -0,0 +1,96 @@
+/**
+ * @author Edouard DUPIN
+ * 
+ * @copyright 2015, Edouard DUPIN, all right reserved
+ * 
+ * @license APACHE v2.0 (see license file)
+ */
+
+#include "debug.h"
+#include "InterfaceInput.h"
+#include <ros/ros.h>
+#include <audio_msg/AudioBuffer.h>
+#include <boost/thread.hpp>
+
+appl::InterfaceInput::InterfaceInput(std11::shared_ptr<river::Manager> _manager,
+                                     const std::string& _input,
+                                     const std::string& _publisher,
+                                     bool _feedback) :
+  m_manager(_manager) {
+	ros::NodeHandle nodeHandlePrivate("~");
+	m_stream = nodeHandlePrivate.advertise<audio_msg::AudioBuffer>(_publisher,
+	                                                               100,
+	                                                               boost::bind(&InterfaceInput::onConnect, this, _1),
+	                                                               boost::bind(&InterfaceInput::onDisConnect, this, _1));
+	//Set stereo output:
+	std::vector<audio::channel> channelMap;
+	channelMap.push_back(audio::channel_frontLeft);
+	channelMap.push_back(audio::channel_frontRight);
+	if (_feedback == false) {
+		m_interface = m_manager->createInput(48000,
+		                                     channelMap,
+		                                     audio::format_int16,
+		                                     _input);
+	} else {
+		m_interface = m_manager->createFeedback(48000,
+		                                        channelMap,
+		                                        audio::format_int16,
+		                                        _input);
+	}
+	if(m_interface == nullptr) {
+		APPL_ERROR("nullptr interface");
+		return;
+	}
+	// set callback mode ...
+	m_interface->setInputCallback(std11::bind(&InterfaceInput::onDataReceived,
+	                                          this,
+	                                          std11::placeholders::_1,
+	                                          std11::placeholders::_2,
+	                                          std11::placeholders::_3,
+	                                          std11::placeholders::_4,
+	                                          std11::placeholders::_5,
+	                                          std11::placeholders::_6));
+	m_interface->start();
+}
+appl::InterfaceInput::~InterfaceInput() {
+	if(m_interface == nullptr) {
+		APPL_ERROR("nullptr interface");
+		return;
+	}
+	m_interface->stop();
+	m_interface.reset();
+}
+void appl::InterfaceInput::onConnect(const ros::SingleSubscriberPublisher& _pub) {
+	APPL_ERROR("on connect ... " << _pub.getSubscriberName());
+}
+void appl::InterfaceInput::onDisConnect(const ros::SingleSubscriberPublisher& _pub) {
+	APPL_ERROR("on dis-connect ... " << _pub.getSubscriberName());
+}
+void appl::InterfaceInput::onDataReceived(const void* _data,
+                                          const std11::chrono::system_clock::time_point& _time,
+                                          size_t _nbChunk,
+                                          enum audio::format _format,
+                                          uint32_t _frequency,
+                                          const std::vector<audio::channel>& _map) {
+	if (_format != audio::format_int16) {
+		APPL_ERROR("call wrong type ... (need int16_t)");
+	}
+	// check if someone suscribe to the output:
+	if (m_stream.getNumSubscribers() > 0) {
+		audio_msg::AudioBuffer msg;
+		// create the Ros timestamp
+		msg.header.stamp = ros::Time::now();//_time;
+		// set message frequency
+		msg.frequency = _frequency;
+		// set channel map properties
+		msg.channelMap = audio::convertChannel(_map);
+		// Set the format of flow
+		msg.channelFormat = audio::convertFormat(_format);
+		// copy data:
+		msg.data.resize(_nbChunk*_map.size()*audio::getFormatBytes(_format));
+		// copy dat aon the ROS buffer interface
+		memcpy(&msg.data[0], _data, _nbChunk*_map.size()*audio::getFormatBytes(_format));
+		// publish message
+		m_stream.publish(msg);
+	}
+}

audio_core/src/InterfaceInput.h

@@ -0,0 +1,39 @@
+/**
+ * @author Edouard DUPIN
+ * 
+ * @copyright 2015, Edouard DUPIN, all right reserved
+ * 
+ * @license APACHE v2.0 (see license file)
+ */
+
+#ifndef __AUDIO_CORE_INTERFACE_INPUT_H__
+#define __AUDIO_CORE_INTERFACE_INPUT_H__
+
+#include <river/Manager.h>
+
+namespace appl {
+	class InterfaceInput {
+		public:
+			std11::shared_ptr<river::Manager> m_manager;
+			std11::shared_ptr<river::Interface> m_interface;
+			ros::Publisher m_stream;
+			std11::mutex mutex;
+		public:
+			InterfaceInput(std11::shared_ptr<river::Manager> _manager,
+			               const std::string& _input="microphone",
+			               const std::string& _publisher="microphone",
+			               bool _feedback=false);
+			~InterfaceInput();
+		protected:
+			void onConnect(const ros::SingleSubscriberPublisher& _pub);
+			void onDisConnect(const ros::SingleSubscriberPublisher& _pub);
+			void onDataReceived(const void* _data,
+			                    const std11::chrono::system_clock::time_point& _time,
+			                    size_t _nbChunk,
+			                    enum audio::format _format,
+			                    uint32_t _frequency,
+			                    const std::vector<audio::channel>& _map);
+	};
+}
+
+#endif

audio_core/src/InterfaceOutput.cpp

@@ -0,0 +1,63 @@
+/**
+ * @author Edouard DUPIN
+ * 
+ * @copyright 2015, Edouard DUPIN, all right reserved
+ * 
+ * @license APACHE v2.0 (see license file)
+ */
+
+#include "debug.h"
+#include "InterfaceOutput.h"
+
+appl::InterfaceOutput::InterfaceOutput(const std::string& _input, const std::string& _publisher) :
+  m_lowLevelStreamName(_input) {
+	ros::NodeHandle nodeHandlePrivate("~");
+	m_stream = nodeHandlePrivate.subscribe<audio_msg::AudioBuffer>(_publisher,
+	                                                               1000,
+	                                                               boost::bind(&InterfaceOutput::onTopicMessage, this, _1));
+	m_timer = nodeHandlePrivate.createTimer(ros::Duration(ros::Rate(4)), boost::bind(&InterfaceOutput::onTimer, this, _1));
+}
+
+appl::InterfaceOutput::~InterfaceOutput() {
+	std11::unique_lock<std::mutex> lock(m_mutex);
+	m_list.clear();
+}
+
+void appl::InterfaceOutput::onTopicMessage(const audio_msg::AudioBuffer::ConstPtr& _msg) {
+	std11::unique_lock<std::mutex> lock(m_mutex);
+	for (size_t iii=0; iii<m_list.size(); ++iii) {
+		if (m_list[iii] == nullptr) {
+			continue;
+		}
+		if (m_list[iii]->getName() == _msg->sourceName) {
+			APPL_VERBOSE("Write data : " << _msg->sourceName);
+			m_list[iii]->onTopicMessage(m_lowLevelStreamName, _msg);
+			return;
+		}
+	}
+	// new interface name:
+	std11::shared_ptr<appl::InterfaceOutputManager> newInterface = std11::make_shared<appl::InterfaceOutputManager>(_msg->sourceName);
+	if (newInterface == nullptr) {
+		APPL_ERROR("can not generate new interface element...");
+		return;
+	}
+	m_list.push_back(newInterface);
+	newInterface->onTopicMessage(m_lowLevelStreamName, _msg);
+}
+
+void appl::InterfaceOutput::onTimer(const ros::TimerEvent& _timer) {
+	std11::unique_lock<std::mutex> lock(m_mutex);
+	std::vector<std11::shared_ptr<appl::InterfaceOutputManager> >::iterator it = m_list.begin();
+	while (it != m_list.end()) {
+		if (*it == nullptr) {
+			it = m_list.erase(it);
+			continue;
+		}
+		if ((*it)->onTimer() == true) {
+			(*it).reset();
+			it = m_list.erase(it);
+			continue;
+		}
+		++it;
+	}
+}

audio_core/src/InterfaceOutput.h

@@ -0,0 +1,31 @@
+/**
+ * @author Edouard DUPIN
+ * 
+ * @copyright 2015, Edouard DUPIN, all right reserved
+ * 
+ * @license APACHE v2.0 (see license file)
+ */
+
+#ifndef __AUDIO_CORE_INTERFACE_OUTPUT_H__
+#define __AUDIO_CORE_INTERFACE_OUTPUT_H__
+
+#include "InterfaceOutputManager.h"
+#include <ros/ros.h>
+#include <audio_msg/AudioBuffer.h>
+
+namespace appl {
+	class InterfaceOutput {
+		public:
+			std::string m_lowLevelStreamName;
+			ros::Subscriber m_stream;
+			ros::Timer m_timer;
+			std11::mutex m_mutex;
+			std::vector<std11::shared_ptr<appl::InterfaceOutputManager> > m_list;
+		public:
+			InterfaceOutput(const std::string& _input="speaker", const std::string& _publisher="speaker");
+			~InterfaceOutput();
+			void onTopicMessage(const audio_msg::AudioBuffer::ConstPtr& _msg);
+			void onTimer(const ros::TimerEvent& _timer);
+	};
+}
+#endif

audio_core/src/InterfaceOutputElement.cpp

@@ -0,0 +1,56 @@
+/**
+ * @author Edouard DUPIN
+ * 
+ * @copyright 2015, Edouard DUPIN, all right reserved
+ * 
+ * @license APACHE v2.0 (see license file)
+ */
+
+#include "debug.h"
+#include "InterfaceOutputElement.h"
+
+appl::InterfaceOutputElement::InterfaceOutputElement(const std11::shared_ptr<river::Manager>& _manager, int32_t _id) :
+  m_id(_id),
+  m_nbConsecutiveUnderflow(0),
+  m_manager(_manager) {
+	APPL_INFO("Create interface");
+}
+
+appl::InterfaceOutputElement::~InterfaceOutputElement() {
+	std11::unique_lock<std::mutex> lock(m_mutex);
+	APPL_INFO("Remove interfaces (start)");
+	m_interface->stop();
+	m_interface.reset();
+	m_manager.reset();
+	APPL_INFO("Remove interfaces (done)");
+}
+
+void appl::InterfaceOutputElement::onTopicMessage(const std::string& _streamName, const audio_msg::AudioBuffer::ConstPtr& _msg) {
+	std11::unique_lock<std::mutex> lock(m_mutex);
+	if (m_interface != nullptr) {
+		APPL_VERBOSE("Write data : " << m_id << " size= " << _msg->data.size()/m_interface->getInterfaceFormat().getChunkSize());
+		m_interface->write(&_msg->data[0], _msg->data.size()/m_interface->getInterfaceFormat().getChunkSize());
+		m_nbConsecutiveUnderflow = 0;
+		return;
+	}
+	audio::format format = audio::convertFormat(_msg->channelFormat);
+	std::vector<enum audio::channel> map = audio::convertChannel(_msg->channelMap);
+	// no interface found => create a new one
+	m_interface = m_manager->createOutput(_msg->frequency,
+	                                      map,
+	                                      format,
+	                                      _streamName);
+	if(m_interface == nullptr) {
+		APPL_ERROR("nullptr interface");
+		return;
+	}
+	m_interface->setReadwrite();
+	m_interface->setStatusFunction(std11::bind(&InterfaceOutputElement::onStatus, this, std11::placeholders::_1, std11::placeholders::_2, _msg->sourceId));
+	m_interface->start();
+	m_interface->write(&_msg->data[0], _msg->data.size()/m_interface->getInterfaceFormat().getChunkSize());
+}
+
+void appl::InterfaceOutputElement::onStatus(const std::string& _origin, const std::string& _status, int32_t _iii) {
+	APPL_VERBOSE("status event : " << _origin << " status=" << _status << " on i=" << _iii);
+	m_nbConsecutiveUnderflow++;
+}

audio_core/src/InterfaceOutputElement.h

@@ -0,0 +1,43 @@
+/**
+ * @author Edouard DUPIN
+ * 
+ * @copyright 2015, Edouard DUPIN, all right reserved
+ * 
+ * @license APACHE v2.0 (see license file)
+ */
+
+#ifndef __AUDIO_CORE_INTERFACE_OUTPUT_ELEMENT_H__
+#define __AUDIO_CORE_INTERFACE_OUTPUT_ELEMENT_H__
+
+#include <river/Manager.h>
+#include <river/Interface.h>
+#include <ros/ros.h>
+#include <audio_msg/AudioBuffer.h>
+
+namespace appl {
+	class InterfaceOutputElement {
+		private:
+			int32_t m_id;
+		public:
+			int32_t getId() {
+				return m_id;
+			}
+		private:
+			int32_t m_nbConsecutiveUnderflow;
+		public:
+			int32_t getCountUnderflow() {
+				return m_nbConsecutiveUnderflow;
+			}
+		private:
+			std11::shared_ptr<river::Manager> m_manager;
+			std11::shared_ptr<river::Interface> m_interface;
+			std11::mutex m_mutex;
+		public:
+			InterfaceOutputElement(const std11::shared_ptr<river::Manager>& _manager, int32_t _id);
+			~InterfaceOutputElement();
+			void onTopicMessage(const std::string& _streamName, const audio_msg::AudioBuffer::ConstPtr& _msg);
+			void onStatus(const std::string& _origin, const std::string& _status, int32_t _iii);
+	};
+}
+
+#endif

audio_core/src/InterfaceOutputManager.cpp

@@ -0,0 +1,70 @@
+/**
+ * @author Edouard DUPIN
+ * 
+ * @copyright 2015, Edouard DUPIN, all right reserved
+ * 
+ * @license APACHE v2.0 (see license file)
+ */
+
+#include "debug.h"
+#include "InterfaceOutputManager.h"
+
+appl::InterfaceOutputManager::InterfaceOutputManager(const std::string& _name) :
+  m_name(_name) {
+	m_manager = river::Manager::create(m_name);
+	APPL_INFO("Create Manager : " << m_name);
+}
+
+appl::InterfaceOutputManager::~InterfaceOutputManager() {
+	APPL_INFO("Remove Manager : " << m_name);
+	std11::unique_lock<std::mutex> lock(m_mutex);
+	APPL_INFO("Clean list");
+	m_elementList.clear();
+	m_manager.reset();
+	APPL_INFO("All is done ...");
+}
+
+void appl::InterfaceOutputManager::onTopicMessage(const std::string& _streamName, const audio_msg::AudioBuffer::ConstPtr& _msg) {
+	std11::unique_lock<std::mutex> lock(m_mutex);
+	for (size_t iii=0; iii<m_elementList.size(); ++iii) {
+		if (m_elementList[iii] == nullptr) {
+			continue;
+		}
+		if(m_elementList[iii]->getId() == _msg->sourceId) {
+			m_elementList[iii]->onTopicMessage(_streamName, _msg);
+			return;
+		}
+	}
+	// no interface found => create a new one
+	std11::shared_ptr<appl::InterfaceOutputElement> interface = std11::make_shared<appl::InterfaceOutputElement>(m_manager, _msg->sourceId);
+	if(interface == nullptr) {
+		APPL_ERROR("nullptr interface");
+		return;
+	}
+	m_elementList.push_back(interface);
+	interface->onTopicMessage(_streamName, _msg);
+	m_manager->generateDotAll("myDot.dot");
+}
+
+bool appl::InterfaceOutputManager::onTimer() {
+	std::vector<std11::shared_ptr<appl::InterfaceOutputElement> >::iterator it = m_elementList.begin();
+	bool oneElementRemoved = false;
+	while (it != m_elementList.end()) {
+		if (*it == nullptr) {
+			it = m_elementList.erase(it);
+			continue;
+		}
+		if ((*it)->getCountUnderflow() >= 50) {
+			(*it).reset();
+			oneElementRemoved = true;
+			it = m_elementList.erase(it);
+			continue;
+		}
+		++it;
+	}
+	if (oneElementRemoved == true) {
+		m_manager->generateDotAll("myDot.dot");
+	}
+	// return remove ...
+	return m_elementList.size() == 0;
+}

audio_core/src/InterfaceOutputManager.h

@@ -0,0 +1,36 @@
+/**
+ * @author Edouard DUPIN
+ * 
+ * @copyright 2015, Edouard DUPIN, all right reserved
+ * 
+ * @license APACHE v2.0 (see license file)
+ */
+
+#ifndef __AUDIO_CORE_INTERFACE_OUTPUT_MANAGER_H__
+#define __AUDIO_CORE_INTERFACE_OUTPUT_MANAGER_H__
+
+#include "InterfaceOutputElement.h"
+#include <ros/ros.h>
+#include <audio_msg/AudioBuffer.h>
+
+namespace appl {
+	class InterfaceOutputManager {
+		private:
+			std::string m_name;
+		public:
+			const std::string& getName() {
+				return m_name;
+			}
+		private:
+			std11::shared_ptr<river::Manager> m_manager;
+			std::vector<std11::shared_ptr<appl::InterfaceOutputElement> > m_elementList;
+			std11::mutex m_mutex;
+		public:
+			InterfaceOutputManager(const std::string& _name);
+			~InterfaceOutputManager();
+			void onTopicMessage(const std::string& _streamName, const audio_msg::AudioBuffer::ConstPtr& _msg);
+			bool onTimer();
+	};
+}
+#endif
+

audio_core/src/main.cpp

@@ -16,285 +16,9 @@
 
 static std11::mutex mutex;
 
-class InterfaceInput {
-	public:
-		std11::shared_ptr<river::Manager> m_manager;
-		std11::shared_ptr<river::Interface> m_interface;
-		ros::Publisher m_stream;
-		std11::mutex mutex;
-	public:
-		InterfaceInput(std11::shared_ptr<river::Manager> _manager, const std::string& _input="microphone", const std::string& _publisher="microphone", bool _feedback=false) :
-		  m_manager(_manager) {
-			ros::NodeHandle nodeHandlePrivate("~");
-			m_stream = nodeHandlePrivate.advertise<audio_msg::AudioBuffer>(_publisher,
-			                                                               100,
-			                                                               boost::bind(&InterfaceInput::onConnect, this, _1),
-			                                                               boost::bind(&InterfaceInput::onDisConnect, this, _1));
-			//Set stereo output:
-			std::vector<audio::channel> channelMap;
-			channelMap.push_back(audio::channel_frontLeft);
-			channelMap.push_back(audio::channel_frontRight);
-			if (_feedback == false) {
-				m_interface = m_manager->createInput(48000,
-				                                     channelMap,
-				                                     audio::format_int16,
-				                                     _input);
-			} else {
-				m_interface = m_manager->createFeedback(48000,
-				                                        channelMap,
-				                                        audio::format_int16,
-				                                        _input);
-			}
-			if(m_interface == nullptr) {
-				APPL_ERROR("nullptr interface");
-				return;
-			}
-			// set callback mode ...
-			m_interface->setInputCallback(std11::bind(&InterfaceInput::onDataReceived,
-			                                          this,
-			                                          std11::placeholders::_1,
-			                                          std11::placeholders::_2,
-			                                          std11::placeholders::_3,
-			                                          std11::placeholders::_4,
-			                                          std11::placeholders::_5,
-			                                          std11::placeholders::_6));
-			m_interface->start();
-		}
-		~InterfaceInput() {
-			if(m_interface == nullptr) {
-				APPL_ERROR("nullptr interface");
-				return;
-			}
-			m_interface->stop();
-			m_interface.reset();
-		}
-		void onConnect(const ros::SingleSubscriberPublisher& _pub) {
-			APPL_ERROR("on connect ... " << _pub.getSubscriberName());
-		}
-		void onDisConnect(const ros::SingleSubscriberPublisher& _pub) {
-			APPL_ERROR("on dis-connect ... " << _pub.getSubscriberName());
-		}
-		void onDataReceived(const void* _data,
-		                    const std11::chrono::system_clock::time_point& _time,
-		                    size_t _nbChunk,
-		                    enum audio::format _format,
-		                    uint32_t _frequency,
-		                    const std::vector<audio::channel>& _map) {
-			if (_format != audio::format_int16) {
-				APPL_ERROR("call wrong type ... (need int16_t)");
-			}
-			// check if someone suscribe to the output:
-			if (m_stream.getNumSubscribers() > 0) {
-				audio_msg::AudioBuffer msg;
-				// create the Ros timestamp
-				msg.header.stamp = ros::Time::now();//_time;
-				// set message frequency
-				msg.frequency = _frequency;
-				// set channel map properties
-				msg.channelMap = audio::convertChannel(_map);
-				// Set the format of flow
-				msg.channelFormat = audio::convertFormat(_format);
-				// copy data:
-				msg.data.resize(_nbChunk*_map.size()*audio::getFormatBytes(_format));
-				// copy dat aon the ROS buffer interface
-				memcpy(&msg.data[0], _data, _nbChunk*_map.size()*audio::getFormatBytes(_format));
-				// publish message
-				m_stream.publish(msg);
-			}
-		}
-};
 
-class InterfaceOutputStreamElement {
-	private:
-		int32_t m_id;
-	public:
-		int32_t getId() {
-			return m_id;
-		}
-	private:
-		int32_t m_nbConsecutiveUnderflow;
-	public:
-		int32_t getCountUnderflow() {
-			return m_nbConsecutiveUnderflow;
-		}
-	private:
-		std11::shared_ptr<river::Manager> m_manager;
-		std11::shared_ptr<river::Interface> m_interface;
-		std11::mutex m_mutex;
-	public:
-		InterfaceOutputStreamElement(const std11::shared_ptr<river::Manager>& _manager, int32_t _id) :
-		  m_id(_id),
-		  m_nbConsecutiveUnderflow(0),
-		  m_manager(_manager) {
-			APPL_INFO("Create interface");
-		}
-		~InterfaceOutputStreamElement() {
-			std11::unique_lock<std::mutex> lock(m_mutex);
-			APPL_INFO("Remove interfaces (start)");
-			m_interface->stop();
-			m_interface.reset();
-			m_manager.reset();
-			APPL_INFO("Remove interfaces (done)");
-		}
-		void onTopicMessage(const std::string& _streamName, const audio_msg::AudioBuffer::ConstPtr& _msg) {
-			std11::unique_lock<std::mutex> lock(m_mutex);
-			if (m_interface != nullptr) {
-				APPL_VERBOSE("Write data : " << m_id << " size= " << _msg->data.size()/m_interface->getInterfaceFormat().getChunkSize());
-				m_interface->write(&_msg->data[0], _msg->data.size()/m_interface->getInterfaceFormat().getChunkSize());
-				m_nbConsecutiveUnderflow = 0;
-				return;
-			}
-			audio::format format = audio::convertFormat(_msg->channelFormat);
-			std::vector<enum audio::channel> map = audio::convertChannel(_msg->channelMap);
-			// no interface found => create a new one
-			m_interface = m_manager->createOutput(_msg->frequency,
-			                                      map,
-			                                      format,
-			                                      _streamName);
-			if(m_interface == nullptr) {
-				APPL_ERROR("nullptr interface");
-				return;
-			}
-			m_interface->setReadwrite();
-			m_interface->setStatusFunction(std11::bind(&InterfaceOutputStreamElement::onStatus, this, std11::placeholders::_1, std11::placeholders::_2, _msg->sourceId));
-			m_interface->start();
-			m_interface->write(&_msg->data[0], _msg->data.size()/m_interface->getInterfaceFormat().getChunkSize());
-		}
-		void onStatus(const std::string& _origin, const std::string& _status, int32_t _iii) {
-			APPL_VERBOSE("status event : " << _origin << " status=" << _status << " on i=" << _iii);
-			m_nbConsecutiveUnderflow++;
-		}
-};
-
-
-class InterfaceOutputStreamManager {
-	private:
-		std::string m_name;
-	public:
-		const std::string& getName() {
-			return m_name;
-		}
-	private:
-		std11::shared_ptr<river::Manager> m_manager;
-		std::vector<std11::shared_ptr<InterfaceOutputStreamElement> > m_elementList;
-		std11::mutex m_mutex;
-	public:
-		InterfaceOutputStreamManager(const std::string& _name) :
-		  m_name(_name) {
-			m_manager = river::Manager::create(m_name);
-			APPL_INFO("Create Manager : " << m_name);
-		}
-		~InterfaceOutputStreamManager() {
-			APPL_INFO("Remove Manager : " << m_name);
-			std11::unique_lock<std::mutex> lock(m_mutex);
-			APPL_INFO("Clean list");
-			m_elementList.clear();
-			m_manager.reset();
-			APPL_INFO("All is done ...");
-		}
-		void onTopicMessage(const std::string& _streamName, const audio_msg::AudioBuffer::ConstPtr& _msg) {
-			std11::unique_lock<std::mutex> lock(m_mutex);
-			for (size_t iii=0; iii<m_elementList.size(); ++iii) {
-				if (m_elementList[iii] == nullptr) {
-					continue;
-				}
-				if(m_elementList[iii]->getId() == _msg->sourceId) {
-					m_elementList[iii]->onTopicMessage(_streamName, _msg);
-					return;
-				}
-			}
-			// no interface found => create a new one
-			std11::shared_ptr<InterfaceOutputStreamElement> interface = std11::make_shared<InterfaceOutputStreamElement>(m_manager, _msg->sourceId);
-			if(interface == nullptr) {
-				APPL_ERROR("nullptr interface");
-				return;
-			}
-			m_elementList.push_back(interface);
-			interface->onTopicMessage(_streamName, _msg);
-			m_manager->generateDotAll("myDot.dot");
-		}
-		bool onTimer() {
-			std::vector<std11::shared_ptr<InterfaceOutputStreamElement> >::iterator it = m_elementList.begin();
-			bool oneElementRemoved = false;
-			while (it != m_elementList.end()) {
-				if (*it == nullptr) {
-					it = m_elementList.erase(it);
-					continue;
-				}
-				if ((*it)->getCountUnderflow() >= 50) {
-					(*it).reset();
-					oneElementRemoved = true;
-					it = m_elementList.erase(it);
-					continue;
-				}
-				++it;
-			}
-			if (oneElementRemoved == true) {
-				m_manager->generateDotAll("myDot.dot");
-			}
-			// return remove ...
-			return m_elementList.size() == 0;
-		}
-};
-
-class InterfaceOutputStream {
-	public:
-		std::string m_lowLevelStreamName;
-		ros::Subscriber m_stream;
-		ros::Timer m_timer;
-		std11::mutex m_mutex;
-		std::vector<std11::shared_ptr<InterfaceOutputStreamManager> > m_list;
-	public:
-		InterfaceOutputStream(const std::string& _input="speaker", const std::string& _publisher="speaker") :
-		  m_lowLevelStreamName("speaker") {
-			ros::NodeHandle nodeHandlePrivate("~");
-			m_stream = nodeHandlePrivate.subscribe<audio_msg::AudioBuffer>(_publisher,
-			                                                               1000,
-			                                                               boost::bind(&InterfaceOutputStream::onTopicMessage, this, _1));
-			m_timer = nodeHandlePrivate.createTimer(ros::Duration(ros::Rate(4)), boost::bind(&InterfaceOutputStream::onTimer, this, _1));
-		}
-		~InterfaceOutputStream() {
-			std11::unique_lock<std::mutex> lock(m_mutex);
-			m_list.clear();
-		}
-		void onTopicMessage(const audio_msg::AudioBuffer::ConstPtr& _msg) {
-			std11::unique_lock<std::mutex> lock(m_mutex);
-			for (size_t iii=0; iii<m_list.size(); ++iii) {
-				if (m_list[iii] == nullptr) {
-					continue;
-				}
-				if (m_list[iii]->getName() == _msg->sourceName) {
-					APPL_VERBOSE("Write data : " << _msg->sourceName);
-					m_list[iii]->onTopicMessage(m_lowLevelStreamName, _msg);
-					return;
-				}
-			}
-			// new interface name:
-			std11::shared_ptr<InterfaceOutputStreamManager> newInterface = std11::make_shared<InterfaceOutputStreamManager>(_msg->sourceName);
-			if (newInterface == nullptr) {
-				APPL_ERROR("can not generate new interface element...");
-				return;
-			}
-			m_list.push_back(newInterface);
-			newInterface->onTopicMessage(m_lowLevelStreamName, _msg);
-		}
-		void onTimer(const ros::TimerEvent& _timer) {
-			std11::unique_lock<std::mutex> lock(m_mutex);
-			std::vector<std11::shared_ptr<InterfaceOutputStreamManager> >::iterator it = m_list.begin();
-			while (it != m_list.end()) {
-				if (*it == nullptr) {
-					it = m_list.erase(it);
-					continue;
-				}
-				if ((*it)->onTimer() == true) {
-					(*it).reset();
-					it = m_list.erase(it);
-					continue;
-				}
-				++it;
-			}
-		}
-};
+#include "InterfaceInput.h"
+#include "InterfaceOutput.h"
 
 
 std11::shared_ptr<river::Manager> g_manager;
@@ -302,8 +26,8 @@ std11::shared_ptr<river::Manager> g_manager;
 bool f_create(audio_core::create::Request& _req,
               audio_core::create::Response& _res) {
 	/*
-	std11::shared_ptr<InterfaceOutput> newInterface;
-	newInterface = std11::make_shared<InterfaceOutput>(g_manager,
+	std11::shared_ptr<appl::InterfaceOutput> newInterface;
+	newInterface = std11::make_shared<appl::InterfaceOutput>(g_manager,
 	                                                   _req.flowName,
 	                                                   audio::convertFormat(_req.channelFormat),
 	                                                   _req.frequency,
@@ -324,7 +48,7 @@ bool f_create(audio_core::create::Request& _req,
 bool f_remove(audio_core::remove::Request& _req,
               audio_core::remove::Response& _res) {
 	/*
-	std11::shared_ptr<InterfaceOutput> interface;
+	std11::shared_ptr<appl::InterfaceOutput> interface;
 	mutex.lock();
 	for(size_t iii=0; iii<g_listInterafceOut.size(); ++iii) {
 		if (g_listInterafceOut[iii] == nullptr) {
@@ -354,7 +78,7 @@ bool f_remove(audio_core::remove::Request& _req,
 bool f_getBufferTime(audio_core::getBufferTime::Request& _req,
                      audio_core::getBufferTime::Response& _res) {
 	/*
-	std11::shared_ptr<InterfaceOutput> interface;
+	std11::shared_ptr<appl::InterfaceOutput> interface;
 	// reset ouput
 	_res.microseconds = 0;
 	mutex.lock();
@@ -441,11 +165,11 @@ int main(int _argc, char **_argv) {
 	
 	g_manager = river::Manager::create(n.getNamespace());
 	// start publishing of Microphone
-	std11::shared_ptr<InterfaceInput> m_input = std11::make_shared<InterfaceInput>(g_manager, "microphone", "microphone", false);
+	std11::shared_ptr<appl::InterfaceInput> m_input = std11::make_shared<appl::InterfaceInput>(g_manager, "microphone", "microphone", false);
 	// start publishing of Speaker feedback
-	std11::shared_ptr<InterfaceInput> m_feedback = std11::make_shared<InterfaceInput>(g_manager, "speaker", "feedback/speaker", true);
+	std11::shared_ptr<appl::InterfaceInput> m_feedback = std11::make_shared<appl::InterfaceInput>(g_manager, "speaker", "feedback/speaker", true);
 	// create the Stream for output
-	std11::shared_ptr<InterfaceOutputStream> m_speaker = std11::make_shared<InterfaceOutputStream>("speaker", "speaker");
+	std11::shared_ptr<appl::InterfaceOutput> m_speaker = std11::make_shared<appl::InterfaceOutput>("speaker", "speaker");
 	
 	/*
 	 * A count of how many messages we have sent. This is used to create

audio_msg/CMakeLists.txt

@@ -42,6 +42,9 @@ find_package(catkin REQUIRED COMPONENTS
 add_message_files(
   FILES
   AudioBuffer.msg
+  AudioControl.msg
+  AudioStatus.msg
+  AudioVolume.msg
 )
 
 ## Generate added messages and services with any dependencies listed here

audio_msg/msg/AudioControl.msg

@@ -0,0 +1,8 @@
+# timestanp the audio buffer
+Header header
+# source
+string sourceName
+# source id
+int32 sourceId
+# config json
+string config

audio_msg/msg/AudioStatus.msg

@@ -0,0 +1,10 @@
+# timestanp the audio buffer
+Header header
+# source
+string sourceName
+# source id
+int32 sourceId
+# type of status
+string type
+# real status
+string status

audio_msg/msg/AudioVolume.msg

@@ -0,0 +1,6 @@
+# timestanp the audio buffer
+Header header
+# Volume name
+string volume
+# new volume in dB:
+float32 value

test/test_audio_generator/CMakeLists.txt

@@ -36,6 +36,10 @@ add_executable(${PROJECT_NAME}_node
   src/main.cpp
 )
 
+## Add cmake target dependencies of the executable/library
+## as an example, message headers may need to be generated before nodes
+add_dependencies(${PROJECT_NAME}_node audio_msg_generate_messages_cpp)
+
 target_link_libraries(${PROJECT_NAME}_node ${ALSA_LIBRARIES} pthread)
 
 ## Specify libraries to link a library or executable target against

test/test_audio_generator/src/main.cpp

@@ -2,10 +2,10 @@
 #include "audio_msg/AudioBuffer.h"
 #include "audio_core/create.h"
 #include "audio_core/remove.h"
-#include "audio_core/write.h"
 #include "audio_core/getBufferTime.h"
 
 #include <sstream>
+#include <boost/thread/mutex.hpp>
 
 int64_t audioInterface_create(ros::NodeHandle& _n, ros::Time _timeUs, int32_t _freq, const std::vector<uint8_t> _channelMap) {
 	ros::ServiceClient client = _n.serviceClient<audio_core::create>("create");
@@ -48,49 +48,106 @@ uint32_t audioInterface_getBufferTime(ros::NodeHandle& _n, int64_t _uid) {
 	return 0;
 }
 
+static std::string p_channelToPlay = "/audio/speaker";
+static int32_t p_sampleRate = 48000;
+static int32_t p_nbChannels = 2;
+static int32_t p_frequency = 440;
+
+static int32_t baseDataSize = 0;
+static double phase = 0.0;
+std::vector<uint8_t> channelMap;
+static ros::Publisher stream;
+
+ros::Time nextFrame;
+
+void onTimer(const ros::TimerEvent& _timer) {
+	int32_t frameSizeTime = (double(baseDataSize) / (double)p_sampleRate) * 1000000000.0;
+	static boost::mutex mutex;
+	boost::unique_lock<boost::mutex> lock(mutex);
+	if ((nextFrame - ros::Time::now()) > ros::Duration(0, frameSizeTime)) {
+		return;
+	}
+	
+	audio_msg::AudioBuffer msg;
+	// Basic source name is the curant node handle (it is unique)
+	msg.sourceName = ros::NodeHandle("~").getNamespace();
+	msg.sourceId = 0;
+	// create the Ros timestamp
+	msg.header.stamp = nextFrame;
+	// set message frequency
+	msg.frequency = p_sampleRate;
+	// set channel map properties
+	msg.channelMap = channelMap;
+	// Set the format of flow
+	msg.channelFormat = audio_msg::AudioBuffer::FORMAT_INT16;
+	
+	std::vector<int16_t> data;
+	data.resize(baseDataSize*channelMap.size());
+	double baseCycle = 2.0*M_PI/(double)p_sampleRate * (double)p_frequency;
+	for (int32_t iii=0; iii<data.size()/channelMap.size(); iii++) {
+		for (int32_t jjj=0; jjj<channelMap.size(); jjj++) {
+			data[channelMap.size()*iii+jjj] = cos(phase) * 15000;
+		}
+		phase += baseCycle;
+		if (phase >= 2*M_PI) {
+			phase -= 2*M_PI;
+		}
+	}
+	// copy data:
+	msg.data.resize(data.size()*sizeof(int16_t));
+	memcpy(&msg.data[0], &data[0], data.size()*sizeof(int16_t));
+	// publish message
+	stream.publish(msg);
+	
+	nextFrame += ros::Duration(0, frameSizeTime);
+	ROS_INFO_STREAM("next frame  " << nextFrame );
+}
+
+
 void usage() {
-	ROS_INFO("test_audio_generator usage [channel] [sampleRate] [nbChannels] [frequency] [timeToPlay]:");
-	ROS_INFO("	[channel] output to write data");
-	ROS_INFO("	[sampleRate] number of sample per second");
-	ROS_INFO("	[nbChannels] nb channel");
-	ROS_INFO("	[frequency] frequency to generate");
-	ROS_INFO("	[timeToPlay] time to generate signal");
+	ROS_INFO("test_audio_generator usage -c=[channel] -s=[sampleRate] -n=[nbChannels] -f=[frequency] -t=[timeToPlay]:");
+	ROS_INFO("	[channel] output to write data (default /audio/speaker)");
+	ROS_INFO("	[sampleRate] number of sample per second (default 48000)");
+	ROS_INFO("	[nbChannels] nb channel (default 2)");
+	ROS_INFO("	[frequency] frequency to generate (default 440)");
 	ROS_INFO("ex");
 	ROS_INFO("	rosrun test_audio_generator test_audio_generator_node /audio/ 48000 2 440 10");
 	exit (-1);
 }
 
-int main(int argc, char **argv) {
-	if (argc < 6 ) {
-		ROS_ERROR ("not enought argument : %d", argc);
-		usage();
+int main(int _argc, char **_argv) {
+	ros::init(_argc, _argv, "test_audio_generator");
+	
+	for (int32_t iii=0; iii<_argc ; ++iii) {
+		if (strncmp(_argv[iii],"-c=", 3) == 0) {
+			p_channelToPlay = &_argv[iii][3];
+		} else if (strncmp(_argv[iii],"-s=", 3) == 0) {
+			sscanf(&_argv[iii][3], "%d", &p_sampleRate);
+		} else if (strncmp(_argv[iii],"-n=", 3) == 0) {
+			sscanf(&_argv[iii][3], "%d", &p_nbChannels);
+		} else if (strncmp(_argv[iii],"-f=", 3) == 0) {
+			sscanf(&_argv[iii][3], "%d", &p_frequency);
+		} else if (    strcmp(_argv[iii],"-h") == 0
+		            || strcmp(_argv[iii],"--help") == 0) {
+			usage();
+		}
 	}
-	std::string p_channelToPlay = argv[1];
-	int32_t p_sampleRate = atoi(argv[2]);
-	int32_t p_nbChannels = atoi(argv[3]);
-	int32_t p_frequency = atoi(argv[4]);
-	int32_t p_timeToPlay = atoi(argv[5]);
+	ROS_INFO_STREAM("Run with:");
+	ROS_INFO_STREAM("    channel='" << p_channelToPlay << "'");
+	ROS_INFO_STREAM("    sampleRate=" << p_sampleRate << " Hz");
+	ROS_INFO_STREAM("    nbChannels=" << p_nbChannels);
+	ROS_INFO_STREAM("    frequency=" << p_frequency << " Hz");
 	
-	sscanf(argv[2], "%d", &p_sampleRate);
-	sscanf(argv[3], "%d", &p_nbChannels);
-	sscanf(argv[4], "%d", &p_frequency);
-	sscanf(argv[5], "%d", &p_timeToPlay);
-	
-	ROS_INFO("	rosrun test_audio_generator test_audio_generator_node %s %d %d %d %d", p_channelToPlay.c_str(), p_sampleRate, p_nbChannels, p_frequency, p_timeToPlay);
-	
-	ros::init(argc, argv, "test_audio_generator");
-	ros::NodeHandle n;
-	double phase = 0;
+	ros::NodeHandle nodeHandlePrivate("~");
 	// send data:
 	ros::Time timee = ros::Time();
-	int32_t baseDataSize = 0;
 	if (p_sampleRate <= 32000) {
 		baseDataSize = 1024*2;
 	} else {
 		baseDataSize = 1024*8;
 	}
 	
-	std::vector<uint8_t> channelMap;
+	
 	if (p_nbChannels == 1) {
 		channelMap.push_back(audio_msg::AudioBuffer::CHANNEL_FRONT_CENTER);
 	} else if (p_nbChannels == 2) {
@@ -110,50 +167,18 @@ int main(int argc, char **argv) {
 		exit(-1);
 	}
 	// new interface: just published data:
-	ros::Publisher stream;
+	
+	nextFrame = ros::Time::now();
+	
 	ros::NodeHandle nodeHandle;
 	// create the output stream:
-	stream = nodeHandle.advertise<audio_msg::AudioBuffer>(p_channelToPlay, 100);
+	stream = nodeHandle.advertise<audio_msg::AudioBuffer>(p_channelToPlay, 10);
 	
-	audio_msg::AudioBuffer msg;
-	// Basic source name is the curant node handle (it is unique)
-	msg.sourceName = ros::NodeHandle("~").getNamespace();
-	msg.sourceId = 0;
-	// create the Ros timestamp
-	msg.header.stamp = ros::Time::now();
-	// set message frequency
-	msg.frequency = p_sampleRate;
-	// set channel map properties
-	msg.channelMap = channelMap;
-	// Set the format of flow
-	msg.channelFormat = audio_msg::AudioBuffer::FORMAT_INT16;
 	
-	std::vector<int16_t> data;
-	data.resize(baseDataSize*channelMap.size());
-	double baseCycle = 2.0*M_PI/(double)p_sampleRate * (double)p_frequency;
-	int32_t generateTime = (p_timeToPlay * p_sampleRate) / baseDataSize;
-	for (int32_t kkk=0; kkk<generateTime; ++kkk) {
-		for (int32_t iii=0; iii<data.size()/channelMap.size(); iii++) {
-			for (int32_t jjj=0; jjj<channelMap.size(); jjj++) {
-				data[channelMap.size()*iii+jjj] = cos(phase) * 15000;
-			}
-			phase += baseCycle;
-			if (phase >= 2*M_PI) {
-				phase -= 2*M_PI;
-			}
-		}
-		// copy data:
-		msg.data.resize(data.size()*sizeof(int16_t));
-		memcpy(&msg.data[0], &data[0], data.size()*sizeof(int16_t));
-		// publish message
-		stream.publish(msg);
-		
-		int32_t needSleep = (double(data.size()/channelMap.size()) / (double)p_sampleRate) * 1000000.0 * 0.97;
-		if (kkk >= 5) {
-			ROS_INFO_STREAM("need sleep " << needSleep << " µs for " << data.size()/channelMap.size() << " chunks");
-			usleep(needSleep);
-		}
-	}
+	ros::Timer timer = nodeHandlePrivate.createTimer(ros::Duration(ros::Rate(100)), boost::bind(&onTimer, _1));
+	
+	ros::spin();
+	
 	return 0;
 }
 

test/test_audio_recorder/CMakeLists.txt

@@ -35,6 +35,10 @@ add_executable(${PROJECT_NAME}_node
   src/main.cpp
 )
 
+## Add cmake target dependencies of the executable/library
+## as an example, message headers may need to be generated before nodes
+add_dependencies(${PROJECT_NAME}_node audio_msg_generate_messages_cpp)
+
 target_link_libraries(${PROJECT_NAME}_node ${ALSA_LIBRARIES} pthread)
 
 ## Specify libraries to link a library or executable target against