From a9015edb2220a3882af0dc9d0069a93cab8fcb13 Mon Sep 17 00:00:00 2001 From: badaix Date: Sun, 24 Jan 2016 13:55:05 +0100 Subject: [PATCH] PCM reader is configured by URI --- server/pcmreader/pcmReader.cpp | 49 +++++++++++++++++--- server/pcmreader/pcmReader.h | 10 ++++- server/pcmreader/pcmReaderFactory.cpp | 43 +++++++++--------- server/pcmreader/pcmReaderFactory.h | 2 +- server/pcmreader/pipeReader.cpp | 8 ++-- server/pcmreader/pipeReader.h | 2 +- server/snapServer.cpp | 6 +-- server/streamServer.cpp | 65 +++++++++++---------------- server/streamServer.h | 7 ++- 9 files changed, 108 insertions(+), 84 deletions(-) diff --git a/server/pcmreader/pcmReader.cpp b/server/pcmreader/pcmReader.cpp index 7b8914bd..adf069e1 100644 --- a/server/pcmreader/pcmReader.cpp +++ b/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 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); diff --git a/server/pcmreader/pcmReader.h b/server/pcmreader/pcmReader.h index c5d41718..25243c69 100644 --- a/server/pcmreader/pcmReader.h +++ b/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 active_; std::thread readerThread_; PcmListener* pcmListener_; - msg::SampleFormat sampleFormat_; + ReaderUri uri_; + SampleFormat sampleFormat_; size_t pcmReadMs_; std::unique_ptr encoder_; + std::string name_; }; diff --git a/server/pcmreader/pcmReaderFactory.cpp b/server/pcmreader/pcmReaderFactory.cpp index f1581e54..6b843839 100644 --- a/server/pcmreader/pcmReaderFactory.cpp +++ b/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; } diff --git a/server/pcmreader/pcmReaderFactory.h b/server/pcmreader/pcmReaderFactory.h index e2e80404..82dd5955 100644 --- a/server/pcmreader/pcmReaderFactory.h +++ b/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); }; diff --git a/server/pcmreader/pipeReader.cpp b/server/pcmreader/pipeReader.cpp index 084d949b..ad6aaf60 100644 --- a/server/pcmreader/pipeReader.cpp +++ b/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 + "\""); } diff --git a/server/pcmreader/pipeReader.h b/server/pcmreader/pipeReader.h index 13e5c5ab..da8c493a 100644 --- a/server/pcmreader/pipeReader.h +++ b/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: diff --git a/server/snapServer.cpp b/server/snapServer.cpp index c8f7ffef..5e6f95a6 100644 --- a/server/snapServer.cpp +++ b/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 portValue("p", "port", "server port", settings.port, &settings.port); Value controlPortValue("", "controlPort", "Remote control port", settings.controlPort, &settings.controlPort); - Value sampleFormatValue("s", "sampleformat", "sample format", settings.sampleFormat.getFormat()); + Value sampleFormatValue("s", "sampleformat", "sample format", settings.sampleFormat); Value codecValue("c", "codec", "transport codec [flac|ogg|pcm][:options]\nType codec:? to get codec specific options", settings.codec, &settings.codec); Value fifoValue("f", "fifo", "name of the input fifo file", pcmStream, &pcmStream); Implicit 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 services; services.push_back(AvahiService("_snapcast._tcp", settings.port)); services.push_back(AvahiService("_snapcast-jsonrpc._tcp", settings.controlPort)); diff --git a/server/streamServer.cpp b/server/streamServer.cpp index 935fee9d..cfb9804a 100644 --- a/server/streamServer.cpp +++ b/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 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 mlock(mutex_); - sampleFormat_.refersTo = requestMsg.id; - connection->send(&sampleFormat_); - } else if (requestMsg.request == kHeader) { std::unique_lock 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 reader(PcmReaderFactory::createPcmReader(this, streamUri, settings_.sampleFormat, settings_.codec, settings_.pipeReadMs)); + pcmReader_.push_back(reader); + pcmReader_.back()->start(); + } + acceptor_ = make_shared(*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 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(); } diff --git a/server/streamServer.h b/server/streamServer.h index 558fb59c..267f12bd 100644 --- a/server/streamServer.h +++ b/server/streamServer.h @@ -28,7 +28,7 @@ #include #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 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_; + std::vector> pcmReader_; std::set> sessions_; asio::io_service* io_service_; std::shared_ptr acceptor_; StreamServerSettings settings_; - msg::SampleFormat sampleFormat_; Queue> messages_; std::unique_ptr controlServer_; };