PCM reader is configured by URI

server/pcmreader/pcmReader.cpp

@@ -24,6 +24,7 @@
 #include "pcmReader.h"
 #include "../encoder/encoderFactory.h"
 #include "common/utils.h"
+#include "common/compat.h"
 #include "common/snapException.h"
 #include "common/log.h"
 
@@ -35,6 +36,7 @@ ReaderUri::ReaderUri(const std::string& uri)
 {
 // https://en.wikipedia.org/wiki/Uniform_Resource_Identifier
 // scheme:[//[user:password@]host[:port]][/]path[?query][#fragment]
+// would be more elegant with regex. Not yet supported on my dev machine's gcc 4.8 :(
 	size_t pos;
 	this->uri = uri;
 	string tmp(uri);
@@ -44,7 +46,7 @@ ReaderUri::ReaderUri(const std::string& uri)
 		throw invalid_argument("missing ':'");
 	scheme = tmp.substr(0, pos);
 	tmp = tmp.substr(pos + 1);
-//	logD << "scheme: '" << scheme << "' tmp: '" << tmp << "'\n";
+	logE << "scheme: '" << scheme << "' tmp: '" << tmp << "'\n";
 
 	if (tmp.find("//") != 0)
 		throw invalid_argument("missing host separator: '//'");
@@ -54,9 +56,9 @@ ReaderUri::ReaderUri(const std::string& uri)
 	if (pos == string::npos)
 		throw invalid_argument("missing path separator: '/'");
 	host = tmp.substr(0, pos);
-	tmp = tmp.substr(pos + 1);
+	tmp = tmp.substr(pos);
 	path = tmp;
-//	logD << "host: '" << host << "' tmp: '" << tmp << "' path: '" << path << "'\n";
+	logE << "host: '" << host << "' tmp: '" << tmp << "' path: '" << path << "'\n";
 
 	pos = tmp.find('?');
 	if (pos == string::npos)
@@ -65,7 +67,7 @@ ReaderUri::ReaderUri(const std::string& uri)
 	path = tmp.substr(0, pos);
 	tmp = tmp.substr(pos + 1);
 	string queryStr = tmp;
-//	logD << "path: '" << path << "' tmp: '" << tmp << "' query: '" << queryStr << "'\n";
+	logE << "path: '" << path << "' tmp: '" << tmp << "' query: '" << queryStr << "'\n";
 
 	pos = tmp.find('#');
 	if (pos != string::npos)
@@ -73,7 +75,7 @@ ReaderUri::ReaderUri(const std::string& uri)
 		queryStr = tmp.substr(0, pos);
 		tmp = tmp.substr(pos + 1);
 		fragment = tmp;
-//		logD << "query: '" << queryStr << "' fragment: '" << fragment << "' tmp: '" << tmp << "'\n";
+		logE << "query: '" << queryStr << "' fragment: '" << fragment << "' tmp: '" << tmp << "'\n";
 	}
 
 	vector<string> keyValueList = split(queryStr, '&');
@@ -89,10 +91,24 @@ ReaderUri::ReaderUri(const std::string& uri)
 }
 
 
-PcmReader::PcmReader(PcmListener* pcmListener, const msg::SampleFormat& sampleFormat, const std::string& codec, const std::string& fifoName, size_t pcmReadMs) : pcmListener_(pcmListener), sampleFormat_(sampleFormat), pcmReadMs_(pcmReadMs)
+PcmReader::PcmReader(PcmListener* pcmListener, const ReaderUri& uri) : pcmListener_(pcmListener), uri_(uri), pcmReadMs_(20)
 {
 	EncoderFactory encoderFactory;
-	encoder_.reset(encoderFactory.createEncoder(codec));
+ 	if (uri_.query.find("codec") == uri_.query.end())
+		throw SnapException("Stream URI must have a codec");
+	encoder_.reset(encoderFactory.createEncoder(uri_.query["codec"]));
+
+	if (uri_.query.find("name") == uri_.query.end())
+		throw SnapException("Stream URI must have a name");
+	name_ = uri_.query["name"];
+
+ 	if (uri_.query.find("sampleformat") == uri_.query.end())
+		throw SnapException("Stream URI must have a sampleformat");
+	sampleFormat_ = SampleFormat(uri_.query["sampleformat"]);
+	logE << "PcmReader sampleFormat: " << sampleFormat_.getFormat() << "\n";
+
+ 	if (uri_.query.find("buffer_ms") != uri_.query.end())
+		pcmReadMs_ = cpt::stoul(uri_.query["buffer_ms"]);
 }
 
 
@@ -108,8 +124,27 @@ msg::Header* PcmReader::getHeader()
 }
 
 
+const ReaderUri& PcmReader::getUri() const
+{
+	return uri_;
+}
+
+
+const std::string& PcmReader::getName() const
+{
+	return name_;
+}
+
+
+const SampleFormat& PcmReader::getSampleFormat() const
+{
+	return sampleFormat_;
+}
+
+
 void PcmReader::start()
 {
+	logE << "PcmReader start: " << sampleFormat_.getFormat() << "\n";
 	encoder_->init(this, sampleFormat_);
  	active_ = true;
 	readerThread_ = thread(&PcmReader::worker, this);

server/pcmreader/pcmReader.h

@@ -74,7 +74,7 @@ class PcmReader : public EncoderListener
 {
 public:
 	/// ctor. Encoded PCM data is passed to the PcmListener
-	PcmReader(PcmListener* pcmListener, const msg::SampleFormat& sampleFormat, const std::string& codec, const std::string& fifoName, size_t pcmReadMs = 20);
+	PcmReader(PcmListener* pcmListener, const ReaderUri& uri);
 	virtual ~PcmReader();
 
 	virtual void start();
@@ -84,6 +84,10 @@ public:
 	virtual void onChunkEncoded(const Encoder* encoder, msg::PcmChunk* chunk, double duration);
 	virtual msg::Header* getHeader();
 
+	virtual const ReaderUri& getUri() const;
+	virtual const std::string& getName() const;
+	virtual const SampleFormat& getSampleFormat() const;
+
 protected:
 	virtual void worker() = 0;
 	int fd_;
@@ -91,9 +95,11 @@ protected:
 	std::atomic<bool> active_;
 	std::thread readerThread_;
 	PcmListener* pcmListener_;
-	msg::SampleFormat sampleFormat_;
+	ReaderUri uri_;
+	SampleFormat sampleFormat_;
 	size_t pcmReadMs_;
 	std::unique_ptr<Encoder> encoder_;
+	std::string name_;
 };
 
 

server/pcmreader/pcmReaderFactory.cpp

@@ -19,35 +19,32 @@
 #include "common/utils.h"
 #include "pcmReaderFactory.h"
 #include "pipeReader.h"
+#include "common/log.h"
 
 
 using namespace std;
 
 
-PcmReader* PcmReaderFactory::createPcmReader(const std::string& uri) const
+PcmReader* PcmReaderFactory::createPcmReader(PcmListener* pcmListener, const std::string& uri, const std::string& defaultSampleFormat, const std::string& defaultCodec, size_t defaultReadBufferMs)
 {
-	PcmReader* pcmReader = NULL;
-/*
-	std::string codec(codecSettings);
-	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 = new OggEncoder(codecOptions);
-	else if (codec == "pcm")
-		encoder = new PcmEncoder(codecOptions);
-	else if (codec == "flac")
-		encoder = new FlacEncoder(codecOptions);
-	else
-	{
-		cout << "unknown codec: " << codec << "\n";
-		return NULL;
-	}
-*/
-	return pcmReader;
+	ReaderUri readerUri(uri);
+
+	if (readerUri.query.find("sampleformat") == readerUri.query.end())
+		readerUri.query["sampleformat"] = defaultSampleFormat;
+
+	if (readerUri.query.find("codec") == readerUri.query.end())
+		readerUri.query["codec"] = defaultCodec;
+
+	logE << "\nURI: " << readerUri.uri << "\nscheme: " << readerUri.scheme << "\nhost: "
+		<< readerUri.host << "\npath: " << readerUri.path << "\nfragment: " << readerUri.fragment << "\n";
+
+	for (auto kv: readerUri.query)
+		logE << "key: '" << kv.first << "' value: '" << kv.second << "'\n";
+
+	if (readerUri.scheme == "pipe")
+		return new PipeReader(pcmListener, readerUri);//, sampleFormat, codec, pcmReadMs);
+
+	return NULL;
 }
 
 

server/pcmreader/pcmReaderFactory.h

@@ -7,7 +7,7 @@
 class PcmReaderFactory
 {
 public:
-	PcmReader* createPcmReader(const std::string& uri) const;
+	static PcmReader* createPcmReader(PcmListener* pcmListener, const std::string& uri, const std::string& defaultSampleFormat, const std::string& defaultCodec, size_t defaultReadBufferMs = 20);
 };
 
 

server/pcmreader/pipeReader.cpp

@@ -32,13 +32,13 @@ using namespace std;
 
 
 
-PipeReader::PipeReader(PcmListener* pcmListener, const msg::SampleFormat& sampleFormat, const std::string& codec, const std::string& fifoName, size_t pcmReadMs) : PcmReader(pcmListener, sampleFormat, codec, fifoName, pcmReadMs)
+PipeReader::PipeReader(PcmListener* pcmListener, const ReaderUri& uri) : PcmReader(pcmListener, uri)
 {
 	umask(0);
-	mkfifo(fifoName.c_str(), 0666);
-	fd_ = open(fifoName.c_str(), O_RDONLY | O_NONBLOCK);
+	mkfifo(uri_.path.c_str(), 0666);
+	fd_ = open(uri_.path.c_str(), O_RDONLY | O_NONBLOCK);
 	if (fd_ == -1)
-		throw SnapException("failed to open fifo: \"" + fifoName + "\"");
+		throw SnapException("failed to open fifo: \"" + uri_.path + "\"");
 }
 
 

server/pcmreader/pipeReader.h

@@ -33,7 +33,7 @@ class PipeReader : public PcmReader
 {
 public:
 	/// ctor. Encoded PCM data is passed to the PipeListener
-	PipeReader(PcmListener* pcmListener, const msg::SampleFormat& sampleFormat, const std::string& codec, const std::string& fifoName, size_t pcmReadMs = 20);
+	PipeReader(PcmListener* pcmListener, const ReaderUri& uri);
 	virtual ~PipeReader();
 
 protected:

server/snapServer.cpp

@@ -46,14 +46,14 @@ int main(int argc, char* argv[])
 	try
 	{
 		StreamServerSettings settings;
-		std::string pcmStream = "pipe:///tmp/snapfifo";
+		std::string pcmStream = "pipe:///tmp/snapfifo?name=default";
 		int processPriority(-3);
 
 		Switch helpSwitch("h", "help", "produce help message");
 		Switch versionSwitch("v", "version", "show version number");
 		Value<size_t> portValue("p", "port", "server port", settings.port, &settings.port);
 		Value<size_t> controlPortValue("", "controlPort", "Remote control port", settings.controlPort, &settings.controlPort);
-		Value<string> sampleFormatValue("s", "sampleformat", "sample format", settings.sampleFormat.getFormat());
+		Value<string> sampleFormatValue("s", "sampleformat", "sample format", settings.sampleFormat);
 		Value<string> codecValue("c", "codec", "transport codec [flac|ogg|pcm][:options]\nType codec:? to get codec specific options", settings.codec, &settings.codec);
 		Value<string> fifoValue("f", "fifo", "name of the input fifo file", pcmStream, &pcmStream);
 		Implicit<int> daemonOption("d", "daemon", "daemonize\noptional process priority [-20..19]", 0, &processPriority);
@@ -141,7 +141,7 @@ int main(int argc, char* argv[])
 			logS(kLogNotice) << "daemon started" << std::endl;
 		}
 
-		PublishAvahi publishAvahi("SnapCast");
+		PublishAvahi publishAvahi("Snapcast");
 		std::vector<AvahiService> services;
 		services.push_back(AvahiService("_snapcast._tcp", settings.port));
 		services.push_back(AvahiService("_snapcast-jsonrpc._tcp", settings.controlPort));

server/streamServer.cpp

@@ -19,9 +19,7 @@
 #include "json/jsonrpc.h"
 #include "streamServer.h"
 #include "message/time.h"
-#include "message/ack.h"
 #include "message/request.h"
-#include "message/command.h"
 #include "message/hello.h"
 #include "common/log.h"
 #include "config.h"
@@ -32,7 +30,7 @@ using namespace std;
 using json = nlohmann::json;
 
 
-StreamServer::StreamServer(asio::io_service* io_service, const StreamServerSettings& streamServerSettings) : io_service_(io_service), settings_(streamServerSettings), sampleFormat_(streamServerSettings.sampleFormat)
+StreamServer::StreamServer(asio::io_service* io_service, const StreamServerSettings& streamServerSettings) : io_service_(io_service), settings_(streamServerSettings)
 {
 }
 
@@ -78,14 +76,22 @@ void StreamServer::send(const msg::BaseMessage* message)
 
 void StreamServer::onChunkRead(const PcmReader* pcmReader, const msg::PcmChunk* chunk, double duration)
 {
-//	logO << "onChunkRead " << duration << "ms\n";
-	send(chunk);
+	logO << "onChunkRead (" << pcmReader->getName() << "): " << duration << "ms\n";
+	bool isDefaultStream(pcmReader == pcmReader_.front().get());
+
+	std::shared_ptr<const msg::BaseMessage> shared_message(chunk);
+	for (auto s : sessions_)
+	{
+		if (isDefaultStream)//->getName() == "default")
+			s->add(shared_message);
+	}
+//	send(chunk);
 }
 
 
 void StreamServer::onResync(const PcmReader* pcmReader, double ms)
 {
-	logO << "onResync " << ms << "ms\n";
+	logO << "onResync (" << pcmReader->getName() << "): " << ms << "ms\n";
 }
 
 
@@ -263,32 +269,15 @@ void StreamServer::onMessageReceived(StreamSession* connection, const msg::BaseM
 				connection->send(&serverSettings);
 			}
 		}
-		else if (requestMsg.request == kSampleFormat)
-		{
-			std::unique_lock<std::mutex> mlock(mutex_);
-			sampleFormat_.refersTo = requestMsg.id;
-			connection->send(&sampleFormat_);
-		}
 		else if (requestMsg.request == kHeader)
 		{
 			std::unique_lock<std::mutex> mlock(mutex_);
-			msg::Header* headerChunk = pcmReader_->getHeader();
+//TODO: use the correct stream
+			msg::Header* headerChunk = pcmReader_.front()->getHeader();
 			headerChunk->refersTo = requestMsg.id;
 			connection->send(headerChunk);
 		}
 	}
-	else if (baseMessage.type == message_type::kCommand)
-	{
-		msg::Command commandMsg;
-		commandMsg.deserialize(baseMessage, buffer);
-		if (commandMsg.getCommand() == "startStream")
-		{
-			msg::Ack ackMsg;
-			ackMsg.refersTo = commandMsg.id;
-			connection->send(&ackMsg);
-			connection->setStreamActive(true);
-		}
-	}
 	else if (baseMessage.type == message_type::kHello)
 	{
 		msg::Hello helloMsg;
@@ -349,20 +338,16 @@ void StreamServer::handleAccept(socket_ptr socket)
 
 void StreamServer::start()
 {
-	for (auto& s: settings_.pcmStreams)
-	{
-		ReaderUri uri(s);
-		logO << "URI: " << uri.uri << "\nscheme: " << uri.scheme << "\nhost: " << uri.host << "\npath: " << uri.path << "\nfragment: " << uri.fragment << "\n";
-		for (auto kv: uri.query)
-			logD << "key: '" << kv.first << "' value: '" << kv.second << "'\n";
-	}
-
 	controlServer_.reset(new ControlServer(io_service_, settings_.controlPort, this));
 	controlServer_->start();
 
-	settings_.pcmStreams[0] = "/tmp/snapfifo";
-	pcmReader_.reset(new PipeReader(this, settings_.sampleFormat, settings_.codec, settings_.pcmStreams[0], settings_.pipeReadMs));
-	pcmReader_->start();
+	for (auto& streamUri: settings_.pcmStreams)
+	{
+		shared_ptr<PcmReader> reader(PcmReaderFactory::createPcmReader(this, streamUri, settings_.sampleFormat, settings_.codec, settings_.pipeReadMs));
+		pcmReader_.push_back(reader);
+		pcmReader_.back()->start();
+	}
+
 	acceptor_ = make_shared<tcp::acceptor>(*io_service_, tcp::endpoint(tcp::v4(), settings_.port));
 	startAccept();
 }
@@ -372,9 +357,11 @@ void StreamServer::stop()
 {
 	controlServer_->stop();
 	acceptor_->cancel();
-	pcmReader_->stop();
+	for (auto pcmReader: pcmReader_)
+		pcmReader->stop();
+
 	std::unique_lock<std::mutex> mlock(mutex_);
-	for (auto it = sessions_.begin(); it != sessions_.end(); ++it)
-		(*it)->stop();
+	for (auto session: sessions_)//it = sessions_.begin(); it != sessions_.end(); ++it)
+		session->stop();
 }
 

server/streamServer.h

@@ -28,7 +28,7 @@
 #include <mutex>
 
 #include "streamSession.h"
-#include "pcmreader/pipeReader.h"
+#include "pcmreader/pcmReaderFactory.h"
 #include "common/queue.h"
 #include "message/message.h"
 #include "message/header.h"
@@ -57,7 +57,7 @@ struct StreamServerSettings
 	std::vector<std::string> pcmStreams;
 	std::string codec;
 	int32_t bufferMs;
-	msg::SampleFormat sampleFormat;
+	std::string sampleFormat;
 	size_t pipeReadMs;
 };
 
@@ -97,13 +97,12 @@ private:
 	void handleAccept(socket_ptr socket);
 	StreamSession* getStreamSession(const std::string& mac);
 	mutable std::mutex mutex_;
-	std::unique_ptr<PcmReader> pcmReader_;
+	std::vector<std::shared_ptr<PcmReader>> pcmReader_;
 	std::set<std::shared_ptr<StreamSession>> sessions_;
 	asio::io_service* io_service_;
 	std::shared_ptr<tcp::acceptor> acceptor_;
 
 	StreamServerSettings settings_;
-	msg::SampleFormat sampleFormat_;
 	Queue<std::shared_ptr<msg::BaseMessage>> messages_;
 	std::unique_ptr<ControlServer> controlServer_;
 };