Merge cbdaeb3ea9 into 40ad2bac0a

server/streamreader/meta_stream.cpp

@@ -39,33 +39,88 @@ MetaStream::MetaStream(PcmStream::Listener* pcmListener, const std::vector<std::
     : PcmStream(pcmListener, ioc, server_settings, uri), first_read_(true)
 {
     auto path_components = utils::string::split(uri.path, '/');
+
+    for (const auto& stream : streams)
+    {
+        addStream(stream);
+    }
+
+    updateActiveStream();
+}
+
+bool MetaStream::isAllowed(const PcmStream& stream) const
+{
+    auto path_components = utils::string::split(uri_.path, '/');
     for (const auto& component : path_components)
     {
         if (component.empty())
             continue;
 
-        bool found = false;
+        if (component == WILDCARD || stream.getName() == component)
-        for (const auto& stream : streams)
         {
-            if (stream->getName() == component)
+            return true;
-            {
-                streams_.push_back(stream);
-                stream->addListener(this);
-                found = true;
-                break;
-            }
         }
-        if (!found)
-            throw SnapException("Unknown stream: \"" + component + "\"");
     }
 
-    if (streams_.empty())
+    return false;
-        throw SnapException("Meta stream '" + getName() + "' must contain at least one stream");
-
-    active_stream_ = streams_.front();
-    resampler_ = make_unique<Resampler>(active_stream_->getSampleFormat(), sampleFormat_);
 }
 
+void MetaStream::addStream(std::shared_ptr<PcmStream> stream)
+{
+    if (isAllowed(*stream))
+    {
+        stream->addListener(this);
+        streams_.push_back(std::move(stream));
+        updateActiveStream();
+    }
+}
+
+void MetaStream::removeStream(const PcmStream& stream)
+{
+    auto iter = std::find_if(streams_.begin(), streams_.end(), [id = stream.getId()](const auto& s) { return s->getId() == id; });
+    if (iter != streams_.end())
+    {
+        streams_.erase(iter);
+        updateActiveStream();
+    }
+}
+
+void MetaStream::updateActiveStream()
+{
+    auto compareStreamOrder = [this](const std::shared_ptr<PcmStream>& first, const std::shared_ptr<PcmStream>& second)
+    {
+        if (first->getName() == second->getName())
+            return false;
+
+        auto path_components = utils::string::split(uri_.path, '/');
+        for (const auto& component : path_components)
+        {
+            if (component == first->getName())
+                return true;
+            if (component == second->getName())
+                return false;
+            if (component == WILDCARD)
+                return false;
+        }
+        return false;
+    };
+
+    std::lock_guard<std::recursive_mutex> lock(active_mutex_);
+    if (!streams_.empty())
+    {
+        auto new_active = std::min_element(streams_.begin(), streams_.end(), compareStreamOrder);
+        if (!active_stream_ || active_stream_->getId() != ((*new_active)->getId()))
+        {
+            active_stream_ = *streams_.begin();
+            resampler_ = make_unique<Resampler>(active_stream_->getSampleFormat(), sampleFormat_);
+        }
+    }
+    else
+    {
+        active_stream_ = nullptr;
+        resampler_ = nullptr;
+    }
+}
 
 MetaStream::~MetaStream()
 {
@@ -133,7 +188,8 @@ void MetaStream::onStateChanged(const PcmStream* pcmStream, ReaderState state)
         }
     }
 
-    switch_stream(streams_.front());
+    if (!streams_.empty())
+        switch_stream(*streams_.begin());
     setState(ReaderState::kIdle);
 }
 
@@ -212,31 +268,36 @@ void MetaStream::onResync(const PcmStream* pcmStream, double ms)
 void MetaStream::setShuffle(bool shuffle, ResultHandler handler)
 {
     std::lock_guard<std::recursive_mutex> lock(mutex_);
-    active_stream_->setShuffle(shuffle, std::move(handler));
+    if (active_stream_)
+        active_stream_->setShuffle(shuffle, std::move(handler));
 }
 
 void MetaStream::setLoopStatus(LoopStatus status, ResultHandler handler)
 {
     std::lock_guard<std::recursive_mutex> lock(mutex_);
-    active_stream_->setLoopStatus(status, std::move(handler));
+    if (active_stream_)
+        active_stream_->setLoopStatus(status, std::move(handler));
 }
 
 void MetaStream::setVolume(uint16_t volume, ResultHandler handler)
 {
     std::lock_guard<std::recursive_mutex> lock(mutex_);
-    active_stream_->setVolume(volume, std::move(handler));
+    if (active_stream_)
+        active_stream_->setVolume(volume, std::move(handler));
 }
 
 void MetaStream::setMute(bool mute, ResultHandler handler)
 {
     std::lock_guard<std::recursive_mutex> lock(mutex_);
-    active_stream_->setMute(mute, std::move(handler));
+    if (active_stream_)
+        active_stream_->setMute(mute, std::move(handler));
 }
 
 void MetaStream::setRate(float rate, ResultHandler handler)
 {
     std::lock_guard<std::recursive_mutex> lock(mutex_);
-    active_stream_->setRate(rate, std::move(handler));
+    if (active_stream_)
+        active_stream_->setRate(rate, std::move(handler));
 }
 
 
@@ -244,54 +305,63 @@ void MetaStream::setRate(float rate, ResultHandler handler)
 void MetaStream::setPosition(std::chrono::milliseconds position, ResultHandler handler)
 {
     std::lock_guard<std::recursive_mutex> lock(mutex_);
-    active_stream_->setPosition(position, std::move(handler));
+    if (active_stream_)
+        active_stream_->setPosition(position, std::move(handler));
 }
 
 void MetaStream::seek(std::chrono::milliseconds offset, ResultHandler handler)
 {
     std::lock_guard<std::recursive_mutex> lock(mutex_);
-    active_stream_->seek(offset, std::move(handler));
+    if (active_stream_)
+        active_stream_->seek(offset, std::move(handler));
 }
 
 void MetaStream::next(ResultHandler handler)
 {
     std::lock_guard<std::recursive_mutex> lock(mutex_);
-    active_stream_->next(std::move(handler));
+    if (active_stream_)
+        active_stream_->next(std::move(handler));
 }
 
 void MetaStream::previous(ResultHandler handler)
 {
     std::lock_guard<std::recursive_mutex> lock(mutex_);
-    active_stream_->previous(std::move(handler));
+    if (active_stream_)
+        active_stream_->previous(std::move(handler));
 }
 
 void MetaStream::pause(ResultHandler handler)
 {
     std::lock_guard<std::recursive_mutex> lock(mutex_);
-    active_stream_->pause(std::move(handler));
+    if (active_stream_)
+        active_stream_->pause(std::move(handler));
 }
 
 void MetaStream::playPause(ResultHandler handler)
 {
     LOG(DEBUG, LOG_TAG) << "PlayPause\n";
     std::lock_guard<std::recursive_mutex> lock(mutex_);
-    if (active_stream_->getState() == ReaderState::kIdle)
+    if (active_stream_)
-        play(handler);
+    {
-    else
+        if (active_stream_->getState() == ReaderState::kIdle)
-        active_stream_->playPause(std::move(handler));
+            play(handler);
+        else
+            active_stream_->playPause(std::move(handler));
+    }
 }
 
 void MetaStream::stop(ResultHandler handler)
 {
     std::lock_guard<std::recursive_mutex> lock(mutex_);
-    active_stream_->stop(std::move(handler));
+    if (active_stream_)
+        active_stream_->stop(std::move(handler));
 }
 
 void MetaStream::play(ResultHandler handler)
 {
     LOG(DEBUG, LOG_TAG) << "Play\n";
     std::lock_guard<std::recursive_mutex> lock(mutex_);
-    if ((active_stream_->getProperties().can_play) && (active_stream_->getProperties().playback_status != PlaybackStatus::kPlaying))
+    if ((active_stream_) && (active_stream_->getProperties().can_play) && (active_stream_->getProperties().playback_status != PlaybackStatus::kPlaying))
         return active_stream_->play(std::move(handler));
 
     for (const auto& stream : streams_)
@@ -303,7 +373,8 @@ void MetaStream::play(ResultHandler handler)
     }
 
     // call play on the active stream to get the handler called
-    active_stream_->play(std::move(handler));
+    if (active_stream_)
+        active_stream_->play(std::move(handler));
 }
 
 

server/streamreader/meta_stream.hpp

@@ -40,6 +40,9 @@ namespace streamreader
  */
 class MetaStream : public PcmStream, public PcmStream::Listener
 {
+public:
+    static inline const std::string WILDCARD = "*";
+
 public:
     /// ctor. Encoded PCM data is passed to the PcmStream::Listener
     MetaStream(PcmStream::Listener* pcmListener, const std::vector<std::shared_ptr<PcmStream>>& streams, boost::asio::io_context& ioc,
@@ -49,6 +52,9 @@ public:
     void start() override;
     void stop() override;
 
+    void addStream(std::shared_ptr<PcmStream> stream);
+    void removeStream(const PcmStream& stream);
+
     // Setter for properties
     void setShuffle(bool shuffle, ResultHandler handler) override;
     void setLoopStatus(LoopStatus status, ResultHandler handler) override;
@@ -67,6 +73,9 @@ public:
     void play(ResultHandler handler) override;
 
 protected:
+    bool isAllowed(const PcmStream& stream) const;
+    void updateActiveStream();
+
     /// Implementation of PcmStream::Listener
     void onPropertiesChanged(const PcmStream* pcmStream, const Properties& properties) override;
     void onStateChanged(const PcmStream* pcmStream, ReaderState state) override;

server/streamreader/stream_manager.cpp

@@ -146,6 +146,9 @@ PcmStreamPtr StreamManager::addStream(StreamUri& streamUri)
         {
             if (s->getName() == stream->getName())
                 throw SnapException("Stream with name \"" + stream->getName() + "\" already exists");
+
+            if (auto meta = dynamic_cast<MetaStream*>(s.get()))
+                meta->addStream(stream);
         }
         streams_.push_back(stream);
     }
@@ -161,6 +164,12 @@ void StreamManager::removeStream(const std::string& name)
     {
         (*iter)->stop();
         streams_.erase(iter);
+
+        for (const auto& s : streams_)
+        {
+            if (auto meta = dynamic_cast<MetaStream*>(s.get()))
+                meta->removeStream(**iter);
+        }
     }
 }