Add null encoder for use with meta streams

README.md

@@ -98,6 +98,7 @@ Available stream sources are:
 - [file](doc/configuration.md#file): read PCM audio from a file
 - [process](doc/configuration.md#process): launches a process and reads audio from stdout
 - [tcp](doc/configuration.md#tcp-server): receives audio from a TCP socket, can act as client or server
+- [meta](doc/configuration.md#meta): read and mix audio from other stream sources
 
 ## Test
 

doc/configuration.md

@@ -197,3 +197,14 @@ The output of any audio player that uses alsa can be redirected to Snapcast by u
     [stream]
     stream = alsa://?name=SomeName&sampleformat=48000:16:2&device=hw:0,1,0
     ```
+
+### meta
+
+Read and mix audio from other stream sources
+
+```sh
+meta:///<name of source#1>/<name of source#2>/.../<name of source#N>?name=<name>
+```
+
+Plays audio from the active source with the highest priority, with `source#1` having the highest priority and `source#N` the lowest.  
+Use `codec=null` for stream sources that should only serve as input for meta streams

server/CMakeLists.txt

@@ -12,6 +12,7 @@ set(SERVER_SOURCES
     stream_session_ws.cpp
     encoder/encoder_factory.cpp
     encoder/pcm_encoder.cpp
+    encoder/null_encoder.cpp
     streamreader/base64.cpp
     streamreader/stream_uri.cpp
     streamreader/stream_manager.cpp

server/Makefile

@@ -44,7 +44,7 @@ endif
 
 CXXFLAGS += $(ADD_CFLAGS) -std=c++14 -Wall -Wextra -Wpedantic -Wno-unused-function -DBOOST_ERROR_CODE_HEADER_ONLY -DHAS_FLAC -DHAS_OGG -DHAS_VORBIS -DHAS_VORBIS_ENC -DHAS_OPUS -DVERSION=\"$(VERSION)\" -I. -I.. -I../common
 LDFLAGS  += $(ADD_LDFLAGS) -lvorbis -lvorbisenc -logg -lFLAC -lopus -lsoxr
-OBJ       = snapserver.o server.o config.o control_server.o control_session_tcp.o control_session_http.o control_session_ws.o stream_server.o stream_session.o stream_session_tcp.o stream_session_ws.o streamreader/stream_uri.o streamreader/base64.o streamreader/stream_manager.o streamreader/pcm_stream.o streamreader/posix_stream.o streamreader/pipe_stream.o streamreader/file_stream.o streamreader/tcp_stream.o streamreader/process_stream.o streamreader/airplay_stream.o streamreader/meta_stream.o streamreader/librespot_stream.o streamreader/watchdog.o encoder/encoder_factory.o encoder/flac_encoder.o encoder/opus_encoder.o encoder/pcm_encoder.o encoder/ogg_encoder.o ../common/sample_format.o ../common/resampler.o
+OBJ       = snapserver.o server.o config.o control_server.o control_session_tcp.o control_session_http.o control_session_ws.o stream_server.o stream_session.o stream_session_tcp.o stream_session_ws.o streamreader/stream_uri.o streamreader/base64.o streamreader/stream_manager.o streamreader/pcm_stream.o streamreader/posix_stream.o streamreader/pipe_stream.o streamreader/file_stream.o streamreader/tcp_stream.o streamreader/process_stream.o streamreader/airplay_stream.o streamreader/meta_stream.o streamreader/librespot_stream.o streamreader/watchdog.o encoder/encoder_factory.o encoder/flac_encoder.o encoder/opus_encoder.o encoder/pcm_encoder.o encoder/null_encoder.o encoder/ogg_encoder.o ../common/sample_format.o ../common/resampler.o
 
 ifneq (,$(TARGET))
 CXXFLAGS += -D$(TARGET)

server/encoder/encoder_factory.cpp

@@ -17,6 +17,7 @@
 ***/
 
 #include "encoder_factory.hpp"
+#include "null_encoder.hpp"
 #include "pcm_encoder.hpp"
 #if defined(HAS_OGG) && defined(HAS_VORBIS) && defined(HAS_VORBIS_ENC)
 #include "ogg_encoder.hpp"
@@ -48,6 +49,8 @@ std::unique_ptr<Encoder> EncoderFactory::createEncoder(const std::string& codecS
     }
     if (codec == "pcm")
         return std::make_unique<PcmEncoder>(codecOptions);
+    else if (codec == "null")
+        return std::make_unique<NullEncoder>(codecOptions);
 #if defined(HAS_OGG) && defined(HAS_VORBIS) && defined(HAS_VORBIS_ENC)
     else if (codec == "ogg")
         return std::make_unique<OggEncoder>(codecOptions);

server/encoder/null_encoder.cpp

@@ -0,0 +1,48 @@
+/***
+    This file is part of snapcast
+    Copyright (C) 2014-2020  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 "null_encoder.hpp"
+
+
+namespace encoder
+{
+
+
+NullEncoder::NullEncoder(const std::string& codecOptions) : Encoder(codecOptions)
+{
+    headerChunk_.reset(new msg::CodecHeader("null"));
+}
+
+
+void NullEncoder::encode(const msg::PcmChunk& chunk)
+{
+    std::ignore = chunk;
+}
+
+
+void NullEncoder::initEncoder()
+{
+}
+
+
+std::string NullEncoder::name() const
+{
+    return "null";
+}
+
+} // namespace encoder

server/encoder/null_encoder.hpp

@@ -0,0 +1,39 @@
+/***
+    This file is part of snapcast
+    Copyright (C) 2014-2020  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 NULL_ENCODER_HPP
+#define NULL_ENCODER_HPP
+#include "encoder.hpp"
+
+namespace encoder
+{
+
+class NullEncoder : public Encoder
+{
+public:
+    NullEncoder(const std::string& codecOptions = "");
+    void encode(const msg::PcmChunk& chunk) override;
+    std::string name() const override;
+
+protected:
+    void initEncoder() override;
+};
+
+} // namespace encoder
+
+#endif

server/etc/snapserver.conf

@@ -126,6 +126,7 @@ doc_root = /usr/share/snapserver/snapweb
 # tcp server: tcp://<listen IP, e.g. 127.0.0.1>:<port>?name=<name>[&mode=server]
 # tcp client: tcp://<server IP, e.g. 127.0.0.1>:<port>?name=<name>&mode=client
 # alsa: alsa://?name=<name>&device=<alsa device>
+# meta: meta:///<name of source#1>/<name of source#2>/.../<name of source#N>?name=<name>
 source = pipe:///tmp/snapfifo?name=default
 #source = tcp://127.0.0.1?name=mopidy_tcp
 

server/streamreader/meta_stream.cpp

@@ -29,7 +29,7 @@ namespace streamreader
 {
 
 static constexpr auto LOG_TAG = "MetaStream";
-static constexpr auto kResyncTolerance = 50ms;
+// static constexpr auto kResyncTolerance = 50ms;
 
 
 MetaStream::MetaStream(PcmListener* pcmListener, std::vector<std::shared_ptr<PcmStream>> streams, boost::asio::io_context& ioc, const StreamUri& uri)
@@ -40,7 +40,7 @@ MetaStream::MetaStream(PcmListener* pcmListener, std::vector<std::shared_ptr<Pcm
     {
         if (component.empty())
             continue;
-        LOG(INFO, LOG_TAG) << "Stream: " << component << "\n";
+
         bool found = false;
         for (const auto stream : streams)
         {
@@ -56,9 +56,6 @@ MetaStream::MetaStream(PcmListener* pcmListener, std::vector<std::shared_ptr<Pcm
             throw SnapException("Unknown stream: \"" + component + "\"");
     }
 
-    for (const auto stream : streams_)
-        LOG(INFO, LOG_TAG) << "Stream: " << stream->getName() << ", " << stream->getUri().toString() << "\n";
-
     if (!streams_.empty())
     {
         active_stream_ = streams_.front();
@@ -108,6 +105,7 @@ void MetaStream::onStateChanged(const PcmStream* pcmStream, ReaderState state)
 
             if (active_stream_ != stream)
             {
+                LOG(INFO, LOG_TAG) << "Stream: " << name_ << ", switching active stream: " << (active_stream_?active_stream_->getName():"<null>") << " => " << stream->getName() << "\n";
                 active_stream_ = stream;
                 resampler_ = make_unique<Resampler>(active_stream_->getSampleFormat(), sampleFormat_);
             }
@@ -146,16 +144,16 @@ void MetaStream::onChunkRead(const PcmStream* pcmStream, const msg::PcmChunk& ch
     // Read took longer, wait for the buffer to fill up
     if (next_read < 0ms)
     {
-        if (next_read >= -kResyncTolerance)
+        // if (next_read >= -kResyncTolerance)
-        {
+        // {
-            LOG(INFO, LOG_TAG) << "next read < 0 (" << getName() << "): " << std::chrono::duration_cast<std::chrono::microseconds>(next_read).count() / 1000.
+        //     LOG(INFO, LOG_TAG) << "next read < 0 (" << getName() << "): " << std::chrono::duration_cast<std::chrono::microseconds>(next_read).count() / 1000.
-                               << " ms\n";
+        //                        << " ms\n";
-        }
+        // }
-        else
+        // else
-        {
+        // {
-            resync(-next_read);
+        resync(-next_read);
-            first_read_ = true;
+        first_read_ = true;
-        }
+        // }
     }
 
     if (resampler_ && resampler_->resamplingNeeded())

server/streamreader/pcm_stream.cpp

@@ -50,7 +50,7 @@ PcmStream::PcmStream(PcmListener* pcmListener, boost::asio::io_context& ioc, con
     if (uri_.query.find(kUriSampleFormat) == uri_.query.end())
         throw SnapException("Stream URI must have a sampleformat");
     sampleFormat_ = SampleFormat(uri_.query[kUriSampleFormat]);
-    LOG(INFO, LOG_TAG) << "PcmStream sampleFormat: " << sampleFormat_.toString() << "\n";
+    LOG(INFO, LOG_TAG) << "PcmStream: " << name_ << ", sampleFormat: " << sampleFormat_.toString() << "\n";
 
     if (uri_.query.find(kUriChunkMs) != uri_.query.end())
         chunk_ms_ = cpt::stoul(uri_.query[kUriChunkMs]);
@@ -95,9 +95,15 @@ const SampleFormat& PcmStream::getSampleFormat() const
 }
 
 
+std::string PcmStream::getCodec() const
+{
+    return encoder_->name();
+}
+
+
 void PcmStream::start()
 {
-    LOG(DEBUG, LOG_TAG) << "Start, sampleformat: " << sampleFormat_.toString() << "\n";
+    LOG(DEBUG, LOG_TAG) << "Start: " << name_ << ", sampleformat: " << sampleFormat_.toString() << "\n";
     encoder_->init([this](const encoder::Encoder& encoder, std::shared_ptr<msg::PcmChunk> chunk, double duration) { chunkEncoded(encoder, chunk, duration); },
                    sampleFormat_);
     active_ = true;
@@ -120,7 +126,7 @@ void PcmStream::setState(ReaderState newState)
 {
     if (newState != state_)
     {
-        LOG(INFO, LOG_TAG) << "State changed: " << static_cast<int>(state_) << " => " << static_cast<int>(newState) << "\n";
+        LOG(INFO, LOG_TAG) << "State changed: " << name_ << ", state: " << static_cast<int>(state_) << " => " << static_cast<int>(newState) << "\n";
         state_ = newState;
         for (auto* listener : pcmListeners_)
         {
@@ -212,7 +218,7 @@ void PcmStream::setMeta(const json& jtag)
 {
     meta_.reset(new msg::StreamTags(jtag));
     meta_->msg["STREAM"] = name_;
-    LOG(INFO, LOG_TAG) << "metadata=" << meta_->msg.dump(4) << "\n";
+    LOG(INFO, LOG_TAG) << "Stream: " << name_ << ", metadata=" << meta_->msg.dump(4) << "\n";
 
     // Trigger a stream update
     for (auto* listener : pcmListeners_)

server/streamreader/pcm_stream.hpp

@@ -90,6 +90,7 @@ public:
     virtual const std::string& getName() const;
     virtual const std::string& getId() const;
     virtual const SampleFormat& getSampleFormat() const;
+    virtual std::string getCodec() const;
 
     std::shared_ptr<msg::StreamTags> getMeta() const;
     void setMeta(const json& j);

server/streamreader/stream_manager.cpp

@@ -153,7 +153,12 @@ const PcmStreamPtr StreamManager::getDefaultStream()
     if (streams_.empty())
         return nullptr;
 
-    return streams_.front();
+    for (const auto stream: streams_)
+    {
+        if (stream->getCodec() != "null")
+            return stream;
+    }
+    return nullptr;
 }
 
 
@@ -198,7 +203,8 @@ json StreamManager::toJson() const
 {
     json result = json::array();
     for (auto stream : streams_)
-        result.push_back(stream->toJson());
+        if (stream->getCodec() != "null")
+            result.push_back(stream->toJson());
     return result;
 }