pipe-reader moved into it's own class

server/Makefile

@@ -5,7 +5,7 @@ CC      = /usr/bin/g++
 CFLAGS  = -std=c++0x -Wall -Wno-unused-function -O3 -D_REENTRANT -DVERSION=\"$(VERSION)\" -I..
 LDFLAGS = -lrt -lpthread -lboost_system -lboost_program_options -lvorbis -lvorbisenc -logg -lFLAC -lavahi-client -lavahi-common
 
-OBJ = snapServer.o controlServer.o flacEncoder.o pcmEncoder.o oggEncoder.o serverSession.o publishAvahi.o ../common/log.o ../message/pcmChunk.o ../message/sampleFormat.o
+OBJ = snapServer.o controlServer.o flacEncoder.o pcmEncoder.o oggEncoder.o serverSession.o publishAvahi.o pipeReader.o encoderFactory.o ../common/log.o ../message/pcmChunk.o ../message/sampleFormat.o
 BIN = snapserver
 
 all:	server
@@ -13,7 +13,7 @@ all:	server
 server: $(OBJ)
 	$(CC) $(CFLAGS) -o $(BIN) $(OBJ) $(LDFLAGS)
 	strip $(BIN)
-	
+
 %.o: %.cpp
 	$(CC) $(CFLAGS) -c $< -o $@
 

server/controlServer.cpp

@@ -25,12 +25,13 @@
 #include <iostream>
 
 
-ControlServer::ControlServer(unsigned short port) : port_(port), headerChunk_(NULL), sampleFormat_(NULL)
+ControlServer::ControlServer(const ControlServerSettings& controlServerSettings) : settings_(controlServerSettings), sampleFormat_(controlServerSettings.sampleFormat)
 {
+	serverSettings_.bufferMs = settings_.bufferMs;
 }
 
 
-void ControlServer::send(shared_ptr<msg::BaseMessage> message)
+void ControlServer::send(const msg::BaseMessage* message)
 {
 	std::unique_lock<std::mutex> mlock(mutex_);
 	for (auto it = sessions_.begin(); it != sessions_.end(); )
@@ -38,7 +39,7 @@ void ControlServer::send(shared_ptr<msg::BaseMessage> message)
 		if (!(*it)->active())
 		{
 			logO << "Session inactive. Removing\n";
-			// don't block: remove ServerSession in a thread 
+			// don't block: remove ServerSession in a thread
 			auto func = [](shared_ptr<ServerSession> s)->void{s->stop();};
 			std::thread t(func, *it);
 			t.detach();
@@ -48,19 +49,33 @@ void ControlServer::send(shared_ptr<msg::BaseMessage> message)
 			++it;
 	}
 
+	std::shared_ptr<const msg::BaseMessage> shared_message(message);
 	for (auto s : sessions_)
-		s->add(message);
+		s->add(shared_message);
+}
+
+
+void ControlServer::onChunkRead(const PipeReader* pipeReader, const msg::PcmChunk* chunk)
+{
+//	logO << "onChunkRead " << chunk->duration<chronos::msec>().count() << "ms\n";
+	send(chunk);
+}
+
+
+void ControlServer::onResync(const PipeReader* pipeReader, double ms)
+{
+	logO << "onResync " << ms << "ms\n";
 }
 
 
 void ControlServer::onMessageReceived(ServerSession* connection, const msg::BaseMessage& baseMessage, char* buffer)
 {
-//	logD << "getNextMessage: " << baseMessage.type << ", size: " << baseMessage.size << ", id: " << baseMessage.id << ", refers: " << baseMessage.refersTo << ", sent: " << baseMessage.sent.sec << "," << baseMessage.sent.usec << ", recv: " << baseMessage.received.sec << "," << baseMessage.received.usec << "\n";
+//	logO << "getNextMessage: " << baseMessage.type << ", size: " << baseMessage.size << ", id: " << baseMessage.id << ", refers: " << baseMessage.refersTo << ", sent: " << baseMessage.sent.sec << "," << baseMessage.sent.usec << ", recv: " << baseMessage.received.sec << "," << baseMessage.received.usec << "\n";
 	if (baseMessage.type == message_type::kRequest)
 	{
 		msg::Request requestMsg;
 		requestMsg.deserialize(baseMessage, buffer);
-//		logD << "request: " << requestMsg.request << "\n";
+//		logO << "request: " << requestMsg.request << "\n";
 		if (requestMsg.request == kTime)
 		{
 			msg::Time timeMsg;
@@ -72,20 +87,21 @@ void ControlServer::onMessageReceived(ServerSession* connection, const msg::Base
 		else if (requestMsg.request == kServerSettings)
 		{
 			std::unique_lock<std::mutex> mlock(mutex_);
-			serverSettings_->refersTo = requestMsg.id;
+			serverSettings_.refersTo = requestMsg.id;
-			connection->send(serverSettings_);
+			connection->send(&serverSettings_);
 		}
 		else if (requestMsg.request == kSampleFormat)
 		{
 			std::unique_lock<std::mutex> mlock(mutex_);
-			sampleFormat_->refersTo = requestMsg.id;
+			sampleFormat_.refersTo = requestMsg.id;
-			connection->send(sampleFormat_);
+			connection->send(&sampleFormat_);
 		}
 		else if (requestMsg.request == kHeader)
 		{
 			std::unique_lock<std::mutex> mlock(mutex_);
-			headerChunk_->refersTo = requestMsg.id;
+			msg::Header* headerChunk = pipeReader_->getHeader();
-			connection->send(headerChunk_);
+			headerChunk->refersTo = requestMsg.id;
+			connection->send(headerChunk);
 		}
 	}
 	else if (baseMessage.type == message_type::kCommand)
@@ -105,7 +121,7 @@ void ControlServer::onMessageReceived(ServerSession* connection, const msg::Base
 
 void ControlServer::acceptor()
 {
-	tcp::acceptor a(io_service_, tcp::endpoint(tcp::v4(), port_));
+	tcp::acceptor a(io_service_, tcp::endpoint(tcp::v4(), settings_.port));
 	for (;;)
 	{
 		socket_ptr sock(new tcp::socket(io_service_));
@@ -128,6 +144,8 @@ void ControlServer::acceptor()
 
 void ControlServer::start()
 {
+	pipeReader_ = new PipeReader(this, settings_.sampleFormat, settings_.codec, settings_.fifoName);
+	pipeReader_->start();
 	acceptThread_ = new thread(&ControlServer::acceptor, this);
 }
 
@@ -137,29 +155,3 @@ void ControlServer::stop()
 //		acceptThread->join();
 }
 
-
-void ControlServer::setHeader(msg::Header* header)
-{
-	if (header)
-		headerChunk_ = header;
-}
-
-
-void ControlServer::setFormat(msg::SampleFormat* format)
-{
-	if (format)
-		sampleFormat_ = format;
-}
-
-
-
-void ControlServer::setServerSettings(msg::ServerSettings* settings)
-{
-	if (settings)
-		serverSettings_ = settings;
-}
-
-
-
-
-

server/controlServer.h

@@ -28,6 +28,7 @@
 #include <mutex>
 
 #include "serverSession.h"
+#include "pipeReader.h"
 #include "common/queue.h"
 #include "message/message.h"
 #include "message/header.h"
@@ -40,28 +41,38 @@ typedef std::shared_ptr<tcp::socket> socket_ptr;
 using namespace std;
 
 
-class ControlServer : public MessageReceiver
+struct ControlServerSettings
+{
+	size_t port;
+	std::string fifoName;
+	std::string codec;
+	int32_t bufferMs;
+	msg::SampleFormat sampleFormat;
+};
+
+
+class ControlServer : public MessageReceiver, PipeListener
 {
 public:
-	ControlServer(unsigned short port);
+	ControlServer(const ControlServerSettings& controlServerSettings);
 
 	void start();
 	void stop();
-	void send(shared_ptr<msg::BaseMessage> message);
+	void send(const msg::BaseMessage* message);
 	virtual void onMessageReceived(ServerSession* connection, const msg::BaseMessage& baseMessage, char* buffer);
-	void setHeader(msg::Header* header);
+	virtual void onChunkRead(const PipeReader* pipeReader, const msg::PcmChunk* chunk);
-	void setFormat(msg::SampleFormat* format);
+	virtual void onResync(const PipeReader* pipeReader, double ms);
-	void setServerSettings(msg::ServerSettings* settings);
 
 private:
 	void acceptor();
 	mutable std::mutex mutex_;
+	PipeReader* pipeReader_;
 	set<shared_ptr<ServerSession>> sessions_;
 	boost::asio::io_service io_service_;
-	unsigned short port_;
+
-	msg::Header* headerChunk_;
+	ControlServerSettings settings_;
-	msg::SampleFormat* sampleFormat_;
+	msg::SampleFormat sampleFormat_;
-	msg::ServerSettings* serverSettings_;
+	msg::ServerSettings serverSettings_;
 	thread* acceptThread_;
 	Queue<shared_ptr<msg::BaseMessage>> messages_;
 };

server/pipeReader.cpp

@@ -0,0 +1,145 @@
+/***
+    This file is part of snapcast
+    Copyright (C) 2015  Johannes Pohl
+
+    This program is free software: you can redistribute it and/or modify
+    it under the terms of the GNU General Public License as published by
+    the Free Software Foundation, either version 3 of the License, or
+    (at your option) any later version.
+
+    This program is distributed in the hope that it will be useful,
+    but WITHOUT ANY WARRANTY; without even the implied warranty of
+    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+    GNU General Public License for more details.
+
+    You should have received a copy of the GNU General Public License
+    along with this program.  If not, see <http://www.gnu.org/licenses/>.
+***/
+
+#include <memory>
+#include <sys/stat.h>
+#include <fcntl.h>
+#include <unistd.h>
+
+#include "pipeReader.h"
+#include "encoderFactory.h"
+#include "common/log.h"
+#include "common/snapException.h"
+
+
+using namespace std;
+
+
+
+
+PipeReader::PipeReader(PipeListener* pipeListener, const msg::SampleFormat& sampleFormat, const std::string& codec, const std::string& fifoName) : pipeListener_(pipeListener), sampleFormat_(sampleFormat)
+{
+	umask(0);
+	mkfifo(fifoName.c_str(), 0666);
+	fd_ = open(fifoName.c_str(), O_RDONLY | O_NONBLOCK);
+	if (fd_ == -1)
+		throw SnapException("failed to open fifo: \"" + fifoName + "\"");
+	pcmReadMs_ = 20;
+	EncoderFactory encoderFactory;
+	encoder_.reset(encoderFactory.createEncoder(codec));
+}
+
+
+PipeReader::~PipeReader()
+{
+	stop();
+	close(fd_);
+}
+
+
+msg::Header* PipeReader::getHeader()
+{
+	return encoder_->getHeader();
+}
+
+
+void PipeReader::start()
+{
+	encoder_->init(this, sampleFormat_);
+ 	active_ = true;
+	readerThread_ = thread(&PipeReader::worker, this);
+}
+
+
+void PipeReader::stop()
+{
+	if (active_)
+	{
+		active_ = false;
+		readerThread_.join();
+	}
+}
+
+
+void PipeReader::onChunkEncoded(const Encoder* encoder, msg::PcmChunk* chunk, double duration)
+{
+//	logO << "onChunkEncoded: " << duration << " us\n";
+	if (duration <= 0)
+		return;
+
+	chunk->timestamp.sec = tvEncodedChunk_.tv_sec;
+	chunk->timestamp.usec = tvEncodedChunk_.tv_usec;
+	chronos::addUs(tvEncodedChunk_, duration * 1000);
+	pipeListener_->onChunkRead(this, chunk);
+}
+
+
+void PipeReader::worker()
+{
+	timeval tvChunk;
+	std::unique_ptr<msg::PcmChunk> chunk(new msg::PcmChunk(sampleFormat_, pcmReadMs_));
+
+	while (active_)
+	{
+		gettimeofday(&tvChunk, NULL);
+		tvEncodedChunk_ = tvChunk;
+		long nextTick = chronos::getTickCount();
+		try
+		{
+			while (active_)
+			{
+				chunk->timestamp.sec = tvChunk.tv_sec;
+				chunk->timestamp.usec = tvChunk.tv_usec;
+				int toRead = chunk->payloadSize;
+				int len = 0;
+				do
+				{
+					int count = read(fd_, chunk->payload + len, toRead - len);
+
+					if (count <= 0)
+						usleep(100*1000);
+					else
+						len += count;
+				}
+				while ((len < toRead) && active_);
+
+				encoder_->encode(chunk.get());
+				nextTick += pcmReadMs_;
+				chronos::addUs(tvChunk, pcmReadMs_ * 1000);
+				long currentTick = chronos::getTickCount();
+
+				if (nextTick >= currentTick)
+				{
+//					logO << "sleep: " << nextTick - currentTick << "\n";
+					usleep((nextTick - currentTick) * 1000);
+				}
+				else
+				{
+					gettimeofday(&tvChunk, NULL);
+					tvEncodedChunk_ = tvChunk;
+					pipeListener_->onResync(this, currentTick - nextTick);
+					nextTick = currentTick;
+				}
+			}
+		}
+		catch(const std::exception& e)
+		{
+			std::cerr << "Exception: " << e.what() << std::endl;
+		}
+	}
+}

server/pipeReader.h

@@ -0,0 +1,66 @@
+/***
+    This file is part of snapcast
+    Copyright (C) 2015  Johannes Pohl
+
+    This program is free software: you can redistribute it and/or modify
+    it under the terms of the GNU General Public License as published by
+    the Free Software Foundation, either version 3 of the License, or
+    (at your option) any later version.
+
+    This program is distributed in the hope that it will be useful,
+    but WITHOUT ANY WARRANTY; without even the implied warranty of
+    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+    GNU General Public License for more details.
+
+    You should have received a copy of the GNU General Public License
+    along with this program.  If not, see <http://www.gnu.org/licenses/>.
+***/
+
+#ifndef PIPE_READER_H
+#define PIPE_READER_H
+
+#include <thread>
+#include <atomic>
+#include <string>
+#include "encoder.h"
+#include "message/sampleFormat.h"
+#include "message/header.h"
+
+
+class PipeReader;
+
+class PipeListener
+{
+public:
+	virtual void onChunkRead(const PipeReader* pipeReader, const msg::PcmChunk* chunk) = 0;
+	virtual void onResync(const PipeReader* pipeReader, double ms) = 0;
+};
+
+
+
+class PipeReader : public EncoderListener
+{
+public:
+	PipeReader(PipeListener* pipeListener, const msg::SampleFormat& sampleFormat, const std::string& codec, const std::string& fifoName);
+	virtual ~PipeReader();
+
+	void start();
+	void stop();
+
+	virtual void onChunkEncoded(const Encoder* encoder, msg::PcmChunk* chunk, double duration);
+	msg::Header* getHeader();
+
+protected:
+	void worker();
+	int fd_;
+	size_t pcmReadMs_;
+	timeval tvEncodedChunk_;
+	std::atomic<bool> active_;
+	std::thread readerThread_;
+	PipeListener* pipeListener_;
+	msg::SampleFormat sampleFormat_;
+	std::unique_ptr<Encoder> encoder_;
+};
+
+
+#endif

server/snapServer.cpp

@@ -20,17 +20,15 @@
 #include <chrono>
 #include <memory>
 
+#include <sys/resource.h>
 #include "common/timeDefs.h"
 #include "common/signalHandler.h"
 #include "common/daemon.h"
 #include "common/log.h"
-#include "common/utils.h"
 #include "common/snapException.h"
 #include "message/sampleFormat.h"
 #include "message/message.h"
-#include "pcmEncoder.h"
+#include "encoderFactory.h"
-#include "oggEncoder.h"
-#include "flacEncoder.h"
 #include "controlServer.h"
 #include "publishAvahi.h"
 
@@ -48,24 +46,20 @@ int main(int argc, char* argv[])
 {
 	try
 	{
-		string sampleFormat;
+		ControlServerSettings settings;
-
-		size_t port;
-		string fifoName;
-		string codec;
 		bool runAsDaemon;
-		int32_t bufferMs;
+		string sampleFormat;
 
 		po::options_description desc("Allowed options");
 		desc.add_options()
 		("help,h", "produce help message")
 		("version,v", "show version number")
-		("port,p", po::value<size_t>(&port)->default_value(98765), "server port")
+		("port,p", po::value<size_t>(&settings.port)->default_value(98765), "server port")
 		("sampleformat,s", po::value<string>(&sampleFormat)->default_value("44100:16:2"), "sample format")
-		("codec,c", po::value<string>(&codec)->default_value("flac"), "transport codec [flac|ogg|pcm][:options]. Type codec:? to get codec specific options")
+		("codec,c", po::value<string>(&settings.codec)->default_value("flac"), "transport codec [flac|ogg|pcm][:options]. Type codec:? to get codec specific options")
-		("fifo,f", po::value<string>(&fifoName)->default_value("/tmp/snapfifo"), "name of the input fifo file")
+		("fifo,f", po::value<string>(&settings.fifoName)->default_value("/tmp/snapfifo"), "name of the input fifo file")
 		("daemon,d", po::bool_switch(&runAsDaemon)->default_value(false), "daemonize")
-		("buffer,b", po::value<int32_t>(&bufferMs)->default_value(1000), "buffer [ms]")
+		("buffer,b", po::value<int32_t>(&settings.bufferMs)->default_value(1000), "buffer [ms]")
 		;
 
 		po::variables_map vm;
@@ -81,7 +75,7 @@ int main(int argc, char* argv[])
 		if (vm.count("version"))
 		{
 			cout << "snapserver " << VERSION << "\n"
-				 << "Copyright (C) 2014 BadAix (snapcast@badaix.de).\n"
+				 << "Copyright (C) 2014, 2015 BadAix (snapcast@badaix.de).\n"
 				 << "License GPLv3+: GNU GPL version 3 or later <http://gnu.org/licenses/gpl.html>.\n"
 				 << "This is free software: you are free to change and redistribute it.\n"
 				 << "There is NO WARRANTY, to the extent permitted by law.\n\n"
@@ -89,57 +83,21 @@ int main(int argc, char* argv[])
 			return 1;
 		}
 
+		if (settings.codec.find(":?") != string::npos)
+		{
+			EncoderFactory encoderFactory;
+			std::unique_ptr<Encoder> encoder(encoderFactory.createEncoder(settings.codec));
+			if (encoder)
+			{
+				cout << "Options for codec \"" << encoder->name() << "\":\n"
+					<< "  " << encoder->getAvailableOptions() << "\n"
+					<< "  Default: \"" << encoder->getDefaultOptions() << "\"\n";
+			}
+			return 1;
+		}
+
 		std::clog.rdbuf(new Log("snapserver", LOG_DAEMON));
 
-		msg::SampleFormat format(sampleFormat);
-		std::unique_ptr<Encoder> encoder;
-		
-		std::string codecOptions;
-		if (codec.find(":") != std::string::npos)
-		{
-			codecOptions = trim_copy(codec.substr(codec.find(":") + 1));
-			codec = trim_copy(codec.substr(0, codec.find(":")));
-		}
-		if (codec == "ogg")
-			encoder.reset(new OggEncoder());
-		else if (codec == "pcm")
-			encoder.reset(new PcmEncoder());
-		else if (codec == "flac")
-			encoder.reset(new FlacEncoder());
-		else
-		{
-			cout << "unknown codec: " << codec << "\n";
-			return 1;
-		}
-		if (codecOptions == "?")
-		{
-			cout << "Options for codec \"" << codec << "\":\n"
-				<< "  " << encoder->getAvailableOptions() << "\n"
-				<< "  Default: \"" << encoder->getDefaultOptions() << "\"\n";
-			return 1;			
-		}
-		try
-		{
-			encoder->init(format, codecOptions);
-		}
-		catch (const std::exception& e)
-		{
-			cout << "Error: " << e.what() << "\n";
-			return 1;
-		}
-
-		umask(0);
-		mkfifo(fifoName.c_str(), 0666);
-		int fd = open(fifoName.c_str(), O_RDONLY | O_NONBLOCK);
-		if (fd == -1)
-		{
-			cout << "failed to open fifo: " << fifoName << "\n";
-			return 1;
-		}
-
-		msg::ServerSettings serverSettings;
-		serverSettings.bufferMs = bufferMs;
-
 		signal(SIGHUP, signal_handler);
 		signal(SIGTERM, signal_handler);
 		signal(SIGINT, signal_handler);
@@ -147,74 +105,21 @@ int main(int argc, char* argv[])
 		if (runAsDaemon)
 		{
 			daemonize("/var/run/snapserver.pid");
+			setpriority(PRIO_PROCESS, 0, -5);
 			logS(kLogNotice) << "daemon started." << endl;
 		}
 
-		ControlServer* controlServer = new ControlServer(port);
-		controlServer->setServerSettings(&serverSettings);
-		controlServer->setFormat(&format);
-		controlServer->setHeader(encoder->getHeader());
-		controlServer->start();
-
 		PublishAvahi publishAvahi("SnapCast");
 		std::vector<AvahiService> services;
-		services.push_back(AvahiService("_snapcast._tcp", port));
+		services.push_back(AvahiService("_snapcast._tcp", settings.port));
 		publishAvahi.publish(services);
 
-		timeval tvChunk;
+		settings.sampleFormat = sampleFormat;
-		gettimeofday(&tvChunk, NULL);
+		ControlServer* controlServer = new ControlServer(settings);
-		long nextTick = chronos::getTickCount();
+		controlServer->start();
-		size_t pcmReadMs = 20;
+
-		
 		while (!g_terminated)
-		{
+			usleep(100*1000);
-			try
-			{
-				shared_ptr<msg::PcmChunk> chunk;
-				while (!g_terminated)//cin.good())
-				{
-					chunk.reset(new msg::PcmChunk(sampleFormat, pcmReadMs));
-					int toRead = chunk->payloadSize;
-					int len = 0;
-					do
-					{
-						int count = read(fd, chunk->payload + len, toRead - len);
-
-						if (count <= 0)
-							usleep(100*1000);
-						else
-							len += count;
-					}
-					while ((len < toRead) && !g_terminated);
-
-					chunk->timestamp.sec = tvChunk.tv_sec;
-					chunk->timestamp.usec = tvChunk.tv_usec;
-					double chunkDuration = encoder->encode(chunk.get());
-					if (chunkDuration > 0)
-						controlServer->send(chunk);
-//					logO << chunkDuration << "\n";
-//                    addUs(tvChunk, 1000*chunk->getDuration());
-					nextTick += pcmReadMs;
-					chronos::addUs(tvChunk, chunkDuration * 1000);
-					long currentTick = chronos::getTickCount();
-					if (nextTick > currentTick)
-					{
-						usleep((nextTick - currentTick) * 1000);
-					}	
-					else
-					{
-						gettimeofday(&tvChunk, NULL);
-						nextTick = chronos::getTickCount();
-					}
-				}
-			}
-			catch(const std::exception& e)
-			{
-				std::cerr << "Exception: " << e.what() << std::endl;
-			}
-			close(fd);
-		}
-
 	}
 	catch (const std::exception& e)
 	{

todo.txt

@@ -1,5 +1,2 @@
 client: cannot connect to server when started first?
-server: throw exception in encode init
-server: put only one encoded frame into ota chunks
-client: time sync timeouts over wired LAN
 test 24, 32bit samples