Add control functions to PcmStream

-move json parsing to Server class
-improve error handling with the new ErrorCode object

common/error_code.hpp

@@ -0,0 +1,61 @@
+/***
+    This file is part of snapcast
+    Copyright (C) 2014-2021  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 ERROR_CODE_HPP
+#define ERROR_CODE_HPP
+
+
+#include <optional>
+#include <string>
+#include <system_error>
+
+
+namespace snapcast
+{
+
+struct ErrorCode : public std::error_code
+{
+    ErrorCode() : std::error_code(), detail_(std::nullopt)
+    {
+    }
+
+    ErrorCode(const std::error_code& code) : std::error_code(code), detail_(std::nullopt)
+    {
+    }
+
+    ErrorCode(const std::error_code& code, std::string detail) : std::error_code(code), detail_(std::move(detail))
+    {
+    }
+
+    std::string detailed_message() const
+    {
+        if (detail_.has_value())
+            return message() + ": " + *detail_;
+        return message();
+    }
+
+private:
+    std::optional<std::string> detail_;
+};
+
+
+} // namespace snapcast
+
+
+
+#endif

server/CMakeLists.txt

@@ -14,6 +14,7 @@ set(SERVER_SOURCES
     encoder/pcm_encoder.cpp
     encoder/null_encoder.cpp
     streamreader/base64.cpp
+    streamreader/control_error.cpp
     streamreader/stream_control.cpp
     streamreader/stream_uri.cpp
     streamreader/stream_manager.cpp

server/Makefile

@@ -44,7 +44,7 @@ endif
 
 CXXFLAGS += $(ADD_CFLAGS) -std=c++17 -Wall -Wextra -Wpedantic -Wno-unused-function -DBOOST_ERROR_CODE_HEADER_ONLY -DHAS_FLAC -DHAS_OGG -DHAS_VORBIS -DHAS_VORBIS_ENC -DHAS_OPUS -DHAS_SOXR -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/null_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 streamreader/control_error.o streamreader/stream_control.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/server.cpp

@@ -459,12 +459,64 @@ void Server::processRequest(const jsonrpcpp::request_ptr request, const OnRespon
                 if (stream == nullptr)
                     throw jsonrpcpp::InternalErrorException("Stream not found", request->id());
 
-                stream->control(*request, [request, on_response](const jsonrpcpp::Response& ctrl_response) {
+                if (!request->params().has("command"))
-                    LOG(INFO, LOG_TAG) << "Received response for Stream.Control, id: " << ctrl_response.id() << ", result: " << ctrl_response.result()
+                    throw jsonrpcpp::InvalidParamsException("Parameter 'commmand' is missing", request->id());
-                                       << ", error: " << ctrl_response.error().code() << "\n";
+
-                    auto response = make_shared<jsonrpcpp::Response>(request->id(), ctrl_response.result());
+                auto command = request->params().get<string>("command");
+
+                auto handle_response = [request, on_response, command](const snapcast::ErrorCode& ec) {
+                    LOG(DEBUG, LOG_TAG) << "Response to '" << command << "': " << ec << ", message: " << ec.detailed_message() << ", msg: " << ec.message()
+                                        << ", category: " << ec.category().name() << "\n";
+                    std::shared_ptr<jsonrpcpp::Response> response;
+                    if (ec)
+                        response = make_shared<jsonrpcpp::Response>(request->id(), jsonrpcpp::Error(ec.detailed_message(), ec.value()));
+                    else
+                        response = make_shared<jsonrpcpp::Response>(request->id(), "ok");
                     on_response(response, nullptr);
-                });
+                };
+
+                if (command == "SetPosition")
+                {
+                    if (!request->params().has("params") || !request->params().get("params").contains("Position"))
+                        throw jsonrpcpp::InvalidParamsException("SetPosition requires parameters 'Position'");
+                    auto seconds = request->params().get("params")["Position"].get<float>();
+                    stream->setPosition(std::chrono::milliseconds(static_cast<int>(seconds * 1000)),
+                                        [handle_response](const snapcast::ErrorCode& ec) { handle_response(ec); });
+                }
+                else if (command == "Seek")
+                {
+                    if (!request->params().has("params") || !request->params().get("params").contains("Offset"))
+                        throw jsonrpcpp::InvalidParamsException("Seek requires parameter 'Offset'");
+                    auto offset = request->params().get("params")["Offset"].get<float>();
+                    stream->seek(std::chrono::milliseconds(static_cast<int>(offset * 1000)),
+                                 [handle_response](const snapcast::ErrorCode& ec) { handle_response(ec); });
+                }
+                else if (command == "Next")
+                {
+                    stream->next([handle_response](const snapcast::ErrorCode& ec) { handle_response(ec); });
+                }
+                else if (command == "Previous")
+                {
+                    stream->previous([handle_response](const snapcast::ErrorCode& ec) { handle_response(ec); });
+                }
+                else if (command == "Pause")
+                {
+                    stream->pause([handle_response](const snapcast::ErrorCode& ec) { handle_response(ec); });
+                }
+                else if (command == "PlayPause")
+                {
+                    stream->playPause([handle_response](const snapcast::ErrorCode& ec) { handle_response(ec); });
+                }
+                else if (command == "Stop")
+                {
+                    stream->stop([handle_response](const snapcast::ErrorCode& ec) { handle_response(ec); });
+                }
+                else if (command == "Play")
+                {
+                    stream->play([handle_response](const snapcast::ErrorCode& ec) { handle_response(ec); });
+                }
+                else
+                    throw jsonrpcpp::InvalidParamsException("Command '" + command + "' not supported", request->id());
 
                 return;
             }
@@ -482,12 +534,55 @@ void Server::processRequest(const jsonrpcpp::request_ptr request, const OnRespon
                 if (stream == nullptr)
                     throw jsonrpcpp::InternalErrorException("Stream not found", request->id());
 
-                stream->setProperty(*request, [request, on_response](const jsonrpcpp::Response& props_response) {
+                if (!request->params().has("property"))
-                    LOG(INFO, LOG_TAG) << "Received response for Stream.SetProperty, id: " << props_response.id() << ", result: " << props_response.result()
+                    throw jsonrpcpp::InvalidParamsException("Parameter 'property' is missing", request->id());
-                                       << ", error: " << props_response.error().code() << "\n";
+
-                    auto response = make_shared<jsonrpcpp::Response>(request->id(), props_response.result());
+                if (!request->params().has("value"))
+                    throw jsonrpcpp::InvalidParamsException("Parameter 'value' is missing", request->id());
+
+                auto name = request->params().get<string>("property");
+                auto value = request->params().get("value");
+                LOG(INFO, LOG_TAG) << "Stream '" << streamId << "' set property: " << name << " = " << value << "\n";
+
+                auto handle_response = [request, on_response](const snapcast::ErrorCode& ec) {
+                    LOG(ERROR, LOG_TAG) << "SetShuffle: " << ec << ", message: " << ec.detailed_message() << ", msg: " << ec.message()
+                                        << ", category: " << ec.category().name() << "\n";
+                    std::shared_ptr<jsonrpcpp::Response> response;
+                    if (ec)
+                        response = make_shared<jsonrpcpp::Response>(request->id(), jsonrpcpp::Error(ec.detailed_message(), ec.value()));
+                    else
+                        response = make_shared<jsonrpcpp::Response>(request->id(), "ok");
                     on_response(response, nullptr);
-                });
+                };
+
+                if (name == "loopStatus")
+                {
+                    auto val = value.get<std::string>();
+                    LoopStatus loop_status = loop_status_from_string(val);
+                    if (loop_status == LoopStatus::kUnknown)
+                        throw jsonrpcpp::InvalidParamsException("Value for loopStatus must be one of 'none', 'track', 'playlist'", request->id());
+                    stream->setLoopStatus(loop_status, [handle_response](const snapcast::ErrorCode& ec) { handle_response(ec); });
+                }
+                else if (name == "shuffle")
+                {
+                    if (!value.is_boolean())
+                        throw jsonrpcpp::InvalidParamsException("Value for shuffle must be bool", request->id());
+                    stream->setShuffle(value.get<bool>(), [handle_response](const snapcast::ErrorCode& ec) { handle_response(ec); });
+                }
+                else if (name == "volume")
+                {
+                    if (!value.is_number_integer())
+                        throw jsonrpcpp::InvalidParamsException("Value for volume must be an int", request->id());
+                    stream->setVolume(value.get<int16_t>(), [handle_response](const snapcast::ErrorCode& ec) { handle_response(ec); });
+                }
+                else if (name == "rate")
+                {
+                    if (!value.is_number_float())
+                        throw jsonrpcpp::InvalidParamsException("Value for rate must be float", request->id());
+                    stream->setRate(value.get<float>(), [handle_response](const snapcast::ErrorCode& ec) { handle_response(ec); });
+                }
+                else
+                    throw jsonrpcpp::InvalidParamsException("Property '" + name + "' not supported", request->id());
 
                 return;
             }

server/streamreader/control_error.cpp

@@ -0,0 +1,89 @@
+/***
+    This file is part of snapcast
+    Copyright (C) 2014-2021  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 "control_error.hpp"
+
+namespace snapcast::error::control
+{
+
+namespace detail
+{
+
+struct category : public std::error_category
+{
+public:
+    const char* name() const noexcept override;
+    std::string message(int value) const override;
+};
+
+
+const char* category::name() const noexcept
+{
+    return "control";
+}
+
+std::string category::message(int value) const
+{
+    switch (static_cast<ControlErrc>(value))
+    {
+        case ControlErrc::success:
+            return "Success";
+        case ControlErrc::can_not_control:
+            return "Stream can not be controlled";
+        case ControlErrc::can_go_previous_is_false:
+            return "Stream property can_go_previous is false";
+        case ControlErrc::can_play_is_false:
+            return "Stream property can_play is false";
+        case ControlErrc::can_pause_is_false:
+            return "Stream property can_pause is false";
+        case ControlErrc::can_seek_is_false:
+            return "Stream property can_seek is false";
+        case ControlErrc::can_control_is_false:
+            return "Stream property can_control is false";
+        case ControlErrc::parse_error:
+            return "Parse error";
+        case ControlErrc::invalid_request:
+            return "Invalid request";
+        case ControlErrc::method_not_found:
+            return "Method not found";
+        case ControlErrc::invalid_params:
+            return "Invalid params";
+        case ControlErrc::internal_error:
+            return "Internal error";
+        default:
+            return "Unknown";
+    }
+}
+
+} // namespace detail
+
+const std::error_category& category()
+{
+    // The category singleton
+    static detail::category instance;
+    return instance;
+}
+
+} // namespace snapcast::error::control
+
+std::error_code make_error_code(ControlErrc errc)
+{
+    // Create an error_code with the original mpg123 error value
+    // and the mpg123 error category.
+    return std::error_code(static_cast<int>(errc), snapcast::error::control::category());
+}

server/streamreader/control_error.hpp

@@ -0,0 +1,82 @@
+/***
+    This file is part of snapcast
+    Copyright (C) 2014-2021  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 STREAMREADER_ERROR_HPP
+#define STREAMREADER_ERROR_HPP
+
+
+#include <system_error>
+
+
+// https://www.boost.org/doc/libs/develop/libs/outcome/doc/html/motivation/plug_error_code.html
+// https://akrzemi1.wordpress.com/examples/error_code-example/
+// https://breese.github.io/2017/05/12/customizing-error-codes.html
+// http://blog.think-async.com/2010/04/system-error-support-in-c0x-part-5.html
+
+
+enum class ControlErrc
+{
+    success = 0,
+
+    // Stream can not be controlled
+    can_not_control = 1,
+
+    // Stream property can_go_next is false
+    can_go_next_is_false = 2,
+    // Stream property can_go_previous is false
+    can_go_previous_is_false = 3,
+    // Stream property can_play is false
+    can_play_is_false = 4,
+    // Stream property can_pause is false
+    can_pause_is_false = 5,
+    // Stream property can_seek is false
+    can_seek_is_false = 6,
+    // Stream property can_control is false
+    can_control_is_false = 7,
+
+    // Invalid JSON was received by the server. An error occurred on the server while parsing the JSON text.
+    parse_error = -32700,
+    // The JSON sent is not a valid Request object.
+    invalid_request = -32600,
+    // The method does not exist / is not available.
+    method_not_found = -32601,
+    // Invalid method parameter(s).
+    invalid_params = -32602,
+    // Internal JSON-RPC error.
+    internal_error = -32603
+};
+
+namespace snapcast::error::control
+{
+const std::error_category& category();
+}
+
+
+
+namespace std
+{
+template <>
+struct is_error_code_enum<ControlErrc> : public std::true_type
+{
+};
+} // namespace std
+
+std::error_code make_error_code(ControlErrc);
+
+
+#endif

server/streamreader/meta_stream.cpp

@@ -87,7 +87,7 @@ void MetaStream::stop()
 void MetaStream::onMetadataChanged(const PcmStream* pcmStream, const Metatags& metadata)
 {
     LOG(DEBUG, LOG_TAG) << "onMetadataChanged: " << pcmStream->getName() << "\n";
-    std::lock_guard<std::mutex> lock(mutex_);
+    std::lock_guard<std::recursive_mutex> lock(mutex_);
     if (pcmStream != active_stream_.get())
         return;
     setMetadata(metadata);
@@ -97,7 +97,7 @@ void MetaStream::onMetadataChanged(const PcmStream* pcmStream, const Metatags& m
 void MetaStream::onPropertiesChanged(const PcmStream* pcmStream, const Properties& properties)
 {
     LOG(DEBUG, LOG_TAG) << "onPropertiesChanged: " << pcmStream->getName() << "\n";
-    std::lock_guard<std::mutex> lock(mutex_);
+    std::lock_guard<std::recursive_mutex> lock(mutex_);
     if (pcmStream != active_stream_.get())
         return;
     setProperties(properties);
@@ -107,7 +107,22 @@ void MetaStream::onPropertiesChanged(const PcmStream* pcmStream, const Propertie
 void MetaStream::onStateChanged(const PcmStream* pcmStream, ReaderState state)
 {
     LOG(DEBUG, LOG_TAG) << "onStateChanged: " << pcmStream->getName() << ", state: " << state << "\n";
-    std::lock_guard<std::mutex> lock(mutex_);
+    std::lock_guard<std::recursive_mutex> lock(mutex_);
+
+    if (active_stream_->getProperties().playback_status == PlaybackStatus::kPaused)
+        return;
+
+    auto switch_stream = [this](std::shared_ptr<PcmStream> new_stream) {
+        if (new_stream == active_stream_)
+            return;
+        LOG(INFO, LOG_TAG) << "Stream: " << name_ << ", switching active stream: " << (active_stream_ ? active_stream_->getName() : "<null>") << " => "
+                           << new_stream->getName() << "\n";
+        active_stream_ = new_stream;
+        setMetadata(active_stream_->getMetadata());
+        setProperties(active_stream_->getProperties());
+        resampler_ = make_unique<Resampler>(active_stream_->getSampleFormat(), sampleFormat_);
+    };
+
     for (const auto& stream : streams_)
     {
         if (stream->getState() == ReaderState::kPlaying)
@@ -117,19 +132,15 @@ 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>") << " => "
+                switch_stream(stream);
-                                   << stream->getName() << "\n";
-                active_stream_ = stream;
-                setMetadata(active_stream_->getMetadata());
-                setProperties(active_stream_->getProperties());
-                resampler_ = make_unique<Resampler>(active_stream_->getSampleFormat(), sampleFormat_);
             }
 
             setState(ReaderState::kPlaying);
             return;
         }
     }
-    active_stream_ = streams_.front();
+
+    switch_stream(streams_.front());
     setState(ReaderState::kIdle);
 }
 
@@ -137,7 +148,7 @@ void MetaStream::onStateChanged(const PcmStream* pcmStream, ReaderState state)
 void MetaStream::onChunkRead(const PcmStream* pcmStream, const msg::PcmChunk& chunk)
 {
     // LOG(TRACE, LOG_TAG) << "onChunkRead: " << pcmStream->getName() << ", duration: " << chunk.durationMs() << "\n";
-    std::lock_guard<std::mutex> lock(mutex_);
+    std::lock_guard<std::recursive_mutex> lock(mutex_);
     if (pcmStream != active_stream_.get())
         return;
     // active_stream_->sampleFormat_
@@ -195,24 +206,104 @@ void MetaStream::onChunkEncoded(const PcmStream* pcmStream, std::shared_ptr<msg:
 void MetaStream::onResync(const PcmStream* pcmStream, double ms)
 {
     LOG(DEBUG, LOG_TAG) << "onResync: " << pcmStream->getName() << ", duration: " << ms << " ms\n";
-    std::lock_guard<std::mutex> lock(mutex_);
+    std::lock_guard<std::recursive_mutex> lock(mutex_);
     if (pcmStream != active_stream_.get())
         return;
     resync(std::chrono::nanoseconds(static_cast<int64_t>(ms * 1000000)));
 }
 
 
-void MetaStream::setProperty(const jsonrpcpp::Request& request, const StreamControl::OnResponse& response_handler)
+
+// Setter for properties
+void MetaStream::setShuffle(bool shuffle, ResultHandler handler)
 {
-    std::lock_guard<std::mutex> lock(mutex_);
+    std::lock_guard<std::recursive_mutex> lock(mutex_);
-    active_stream_->setProperty(request, response_handler);
+    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));
+}
+
+void MetaStream::setVolume(uint16_t volume, ResultHandler handler)
+{
+    std::lock_guard<std::recursive_mutex> lock(mutex_);
+    active_stream_->setVolume(volume, 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));
 }
 
 
-void MetaStream::control(const jsonrpcpp::Request& request, const StreamControl::OnResponse& response_handler)
+// Control commands
+void MetaStream::setPosition(std::chrono::milliseconds position, ResultHandler handler)
 {
-    std::lock_guard<std::mutex> lock(mutex_);
+    std::lock_guard<std::recursive_mutex> lock(mutex_);
-    active_stream_->control(request, response_handler);
+    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));
+}
+
+void MetaStream::next(ResultHandler handler)
+{
+    std::lock_guard<std::recursive_mutex> lock(mutex_);
+    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));
+}
+
+void MetaStream::pause(ResultHandler handler)
+{
+    std::lock_guard<std::recursive_mutex> lock(mutex_);
+    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)
+        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));
+}
+
+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))
+        return active_stream_->play(std::move(handler));
+
+    for (const auto& stream : streams_)
+    {
+        if ((stream->getState() == ReaderState::kIdle) && (stream->getProperties().can_play))
+        {
+            return stream->play(std::move(handler));
+        }
+    }
+
+    // call play on the active stream to get the handler called
+    active_stream_->play(std::move(handler));
 }
 
 

server/streamreader/meta_stream.hpp

@@ -44,6 +44,22 @@ public:
     void start() override;
     void stop() override;
 
+    // Setter for properties
+    void setShuffle(bool shuffle, ResultHandler handler) override;
+    void setLoopStatus(LoopStatus status, ResultHandler handler) override;
+    void setVolume(uint16_t volume, ResultHandler handler) override;
+    void setRate(float rate, ResultHandler handler) override;
+
+    // Control commands
+    void setPosition(std::chrono::milliseconds position, ResultHandler handler) override;
+    void seek(std::chrono::milliseconds offset, ResultHandler handler) override;
+    void next(ResultHandler handler) override;
+    void previous(ResultHandler handler) override;
+    void pause(ResultHandler handler) override;
+    void playPause(ResultHandler handler) override;
+    void stop(ResultHandler handler) override;
+    void play(ResultHandler handler) override;
+
 protected:
     /// Implementation of PcmListener
     void onMetadataChanged(const PcmStream* pcmStream, const Metatags& metadata) override;
@@ -54,12 +70,9 @@ protected:
     void onResync(const PcmStream* pcmStream, double ms) override;
 
 protected:
-    void setProperty(const jsonrpcpp::Request& request, const StreamControl::OnResponse& response_handler) override;
-    void control(const jsonrpcpp::Request& request, const StreamControl::OnResponse& response_handler) override;
-
     std::vector<std::shared_ptr<PcmStream>> streams_;
     std::shared_ptr<PcmStream> active_stream_;
-    std::mutex mutex_;
+    std::recursive_mutex mutex_;
     std::unique_ptr<Resampler> resampler_;
     bool first_read_;
     std::chrono::time_point<std::chrono::steady_clock> next_tick_;

server/streamreader/pcm_stream.cpp

@@ -21,9 +21,11 @@
 #include <sys/stat.h>
 
 #include "common/aixlog.hpp"
+#include "common/error_code.hpp"
 #include "common/snap_exception.hpp"
 #include "common/str_compat.hpp"
 #include "common/utils/string_utils.hpp"
+#include "control_error.hpp"
 #include "encoder/encoder_factory.hpp"
 #include "pcm_stream.hpp"
 
@@ -326,126 +328,127 @@ const Properties& PcmStream::getProperties() const
 }
 
 
-void PcmStream::setProperty(const jsonrpcpp::Request& request, const StreamControl::OnResponse& response_handler)
+void PcmStream::setShuffle(bool shuffle, ResultHandler handler)
 {
-    try
+    LOG(DEBUG, LOG_TAG) << "setShuffle: " << shuffle << "\n";
-    {
+    sendRequest("Plugin.Stream.Player.SetProperty", {"shuffle", shuffle}, std::move(handler));
-        if (!request.params().has("property"))
-            throw SnapException("Parameter 'property' is missing");
-
-        if (!request.params().has("value"))
-            throw SnapException("Parameter 'value' is missing");
-
-        auto name = request.params().get("property");
-        auto value = request.params().get("value");
-        LOG(INFO, LOG_TAG) << "Stream '" << getId() << "' set property: " << name << " = " << value << "\n";
-
-        if (name == "loopStatus")
-        {
-            auto val = value.get<std::string>();
-            if ((val != "none") || (val != "track") || (val != "playlist"))
-                throw SnapException("Value for loopStatus must be one of 'none', 'track', 'playlist'");
-        }
-        else if (name == "shuffle")
-        {
-            if (!value.is_boolean())
-                throw SnapException("Value for shuffle must be bool");
-        }
-        else if (name == "volume")
-        {
-            if (!value.is_number_integer())
-                throw SnapException("Value for volume must be an int");
-        }
-        else if (name == "rate")
-        {
-            if (!value.is_number_float())
-                throw SnapException("Value for rate must be float");
-        }
-
-        if (!properties_.can_control)
-            throw SnapException("CanControl is false");
-
-        if (stream_ctrl_)
-        {
-            jsonrpcpp::Request req(++req_id_, "Plugin.Stream.Player.SetProperty", {name, value});
-            stream_ctrl_->command(req, response_handler);
-        }
-    }
-    catch (const std::exception& e)
-    {
-        LOG(WARNING, LOG_TAG) << "Error in setProperty: " << e.what() << '\n';
-        auto error = jsonrpcpp::InvalidParamsException(e.what(), request.id());
-        response_handler(error.to_json());
-    }
 }
 
 
-void PcmStream::control(const jsonrpcpp::Request& request, const StreamControl::OnResponse& response_handler)
+void PcmStream::setLoopStatus(LoopStatus status, ResultHandler handler)
 {
-    try
+    LOG(DEBUG, LOG_TAG) << "setLoopStatus: " << status << "\n";
-    {
+    sendRequest("Plugin.Stream.Player.SetProperty", {"loopStatus", to_string(status)}, std::move(handler));
-        if (!request.params().has("command"))
+}
-            throw SnapException("Parameter 'command' is missing");
 
-        std::string command = request.params().get("command");
+
-        if (command == "SetPosition")
+void PcmStream::setVolume(uint16_t volume, ResultHandler handler)
-        {
+{
-            if (!request.params().has("params") || !request.params().get("params").contains("Position"))
+    LOG(DEBUG, LOG_TAG) << "setVolume: " << volume << "\n";
-                throw SnapException("SetPosition requires parameters 'Position' and optionally 'TrackId'");
+    sendRequest("Plugin.Stream.Player.SetProperty", {"volume", volume}, std::move(handler));
-            if (!properties_.can_seek)
+}
-                throw SnapException("CanSeek is false");
+
-        }
+
-        else if (command == "Seek")
+void PcmStream::setRate(float rate, ResultHandler handler)
-        {
+{
-            if (!request.params().has("params") || !request.params().get("params").contains("Offset"))
+    LOG(DEBUG, LOG_TAG) << "setRate: " << rate << "\n";
-                throw SnapException("Seek requires parameter 'Offset'");
+    sendRequest("Plugin.Stream.Player.SetProperty", {"rate", rate}, std::move(handler));
-            if (!properties_.can_seek)
+}
-                throw SnapException("CanSeek is false");
+
-        }
+
-        else if (command == "Next")
+void PcmStream::setPosition(std::chrono::milliseconds position, ResultHandler handler)
-        {
+{
-            if (!properties_.can_go_next)
+    LOG(DEBUG, LOG_TAG) << "setPosition\n";
-                throw SnapException("CanGoNext is false");
+    if (!properties_.can_seek)
-        }
+        return handler({ControlErrc::can_seek_is_false});
-        else if (command == "Previous")
+    json params;
-        {
+    params["command"] = "SetPosition";
-            if (!properties_.can_go_previous)
+    json j;
-                throw SnapException("CanGoPrevious is false");
+    j["Position"] = position.count() / 1000.f;
-        }
+    params["params"] = j;
-        else if ((command == "Pause") || (command == "PlayPause"))
+    sendRequest("Plugin.Stream.Player.Control", params, std::move(handler));
-        {
+}
-            if (!properties_.can_pause)
+
-                throw SnapException("CanPause is false");
+
-        }
+void PcmStream::seek(std::chrono::milliseconds offset, ResultHandler handler)
-        else if (command == "Stop")
+{
-        {
+    LOG(DEBUG, LOG_TAG) << "seek\n";
-            if (!properties_.can_control)
+    if (!properties_.can_seek)
-                throw SnapException("CanControl is false");
+        return handler({ControlErrc::can_seek_is_false});
-        }
+    json params;
-        else if (command == "Play")
+    params["command"] = "Seek";
-        {
+    json j;
-            if (!properties_.can_play)
+    j["Offset"] = offset.count() / 1000.f;
-                throw SnapException("CanPlay is false");
+    params["params"] = j;
-        }
+    sendRequest("Plugin.Stream.Player.Control", params, std::move(handler));
+}
+
+
+void PcmStream::next(ResultHandler handler)
+{
+    LOG(DEBUG, LOG_TAG) << "next\n";
+    if (!properties_.can_go_next)
+        return handler({ControlErrc::can_go_next_is_false});
+    sendRequest("Plugin.Stream.Player.Control", {"command", "Next"}, std::move(handler));
+}
+
+
+void PcmStream::previous(ResultHandler handler)
+{
+    LOG(DEBUG, LOG_TAG) << "previous\n";
+    if (!properties_.can_go_previous)
+        return handler({ControlErrc::can_go_previous_is_false});
+    sendRequest("Plugin.Stream.Player.Control", {"command", "Previous"}, std::move(handler));
+}
+
+
+void PcmStream::pause(ResultHandler handler)
+{
+    LOG(DEBUG, LOG_TAG) << "pause\n";
+    if (!properties_.can_pause)
+        return handler({ControlErrc::can_pause_is_false});
+    sendRequest("Plugin.Stream.Player.Control", {"command", "Pause"}, std::move(handler));
+}
+
+void PcmStream::sendRequest(const std::string& method, const jsonrpcpp::Parameter& params, ResultHandler handler)
+{
+    if (!stream_ctrl_)
+        return handler({ControlErrc::can_not_control});
+
+    jsonrpcpp::Request req(++req_id_, method, params);
+    stream_ctrl_->command(req, [handler](const jsonrpcpp::Response& response) {
+        if (response.error().code())
+            handler({static_cast<ControlErrc>(response.error().code()), response.error().data()});
         else
-            throw SnapException("Command not supported");
+            handler({ControlErrc::success});
+    });
+}
 
-        LOG(INFO, LOG_TAG) << "Stream '" << getId() << "' received command: '" << command << "', params: '" << request.params().to_json() << "'\n";
+
-        if (stream_ctrl_)
+void PcmStream::playPause(ResultHandler handler)
-        {
+{
-            jsonrpcpp::Parameter params{"command", command};
+    LOG(DEBUG, LOG_TAG) << "playPause\n";
-            if (request.params().has("params"))
+    if (!properties_.can_pause)
-                params.add("params", request.params().get("params"));
+        return handler({ControlErrc::can_play_is_false});
-            jsonrpcpp::Request req(++req_id_, "Plugin.Stream.Player.Control", params);
+    sendRequest("Plugin.Stream.Player.Control", {"command", "PlayPause"}, std::move(handler));
-            stream_ctrl_->command(req, response_handler);
+}
-        }
+
-    }
+
-    catch (const std::exception& e)
+void PcmStream::stop(ResultHandler handler)
-    {
+{
-        LOG(WARNING, LOG_TAG) << "Error in control: " << e.what() << '\n';
+    LOG(DEBUG, LOG_TAG) << "stop\n";
-        auto error = jsonrpcpp::InvalidParamsException(e.what(), request.id());
+    if (!properties_.can_control)
-        response_handler(error.to_json());
+        return handler({ControlErrc::can_control_is_false});
-    }
+    sendRequest("Plugin.Stream.Player.Control", {"command", "Stop"}, std::move(handler));
+}
+
+
+void PcmStream::play(ResultHandler handler)
+{
+    LOG(DEBUG, LOG_TAG) << "play\n";
+    if (!properties_.can_play)
+        return handler({ControlErrc::can_play_is_false});
+    sendRequest("Plugin.Stream.Player.Control", {"command", "Play"}, std::move(handler));
 }
 
 

server/streamreader/pcm_stream.hpp

@@ -28,6 +28,7 @@
 #include <boost/asio/read_until.hpp>
 #include <boost/asio/steady_timer.hpp>
 
+#include "common/error_code.hpp"
 #include "common/json.hpp"
 #include "common/metatags.hpp"
 #include "common/properties.hpp"
@@ -116,6 +117,8 @@ public:
 class PcmStream
 {
 public:
+    using ResultHandler = std::function<void(const snapcast::ErrorCode& ec)>;
+
     /// ctor. Encoded PCM data is passed to the PcmListener
     PcmStream(PcmListener* pcmListener, boost::asio::io_context& ioc, const ServerSettings& server_settings, const StreamUri& uri);
     virtual ~PcmStream();
@@ -134,8 +137,21 @@ public:
     const Metatags& getMetadata() const;
     const Properties& getProperties() const;
 
-    virtual void setProperty(const jsonrpcpp::Request& request, const StreamControl::OnResponse& response_handler);
+    // Setter for properties
-    virtual void control(const jsonrpcpp::Request& request, const StreamControl::OnResponse& response_handler);
+    virtual void setShuffle(bool shuffle, ResultHandler handler);
+    virtual void setLoopStatus(LoopStatus status, ResultHandler handler);
+    virtual void setVolume(uint16_t volume, ResultHandler handler);
+    virtual void setRate(float rate, ResultHandler handler);
+
+    // Control commands
+    virtual void setPosition(std::chrono::milliseconds position, ResultHandler handler);
+    virtual void seek(std::chrono::milliseconds offset, ResultHandler handler);
+    virtual void next(ResultHandler handler);
+    virtual void previous(ResultHandler handler);
+    virtual void pause(ResultHandler handler);
+    virtual void playPause(ResultHandler handler);
+    virtual void stop(ResultHandler handler);
+    virtual void play(ResultHandler handler);
 
     virtual ReaderState getState() const;
     virtual json toJson() const;
@@ -145,10 +161,6 @@ public:
 protected:
     std::atomic<bool> active_;
 
-    void onControlRequest(const jsonrpcpp::Request& request);
-    void onControlNotification(const jsonrpcpp::Notification& notification);
-    void onControlLog(std::string line);
-
     void setState(ReaderState newState);
     void chunkRead(const msg::PcmChunk& chunk);
     void resync(const std::chrono::nanoseconds& duration);
@@ -159,6 +171,16 @@ protected:
 
     void pollProperties();
 
+    // script callbacks
+    /// Request received from control script
+    void onControlRequest(const jsonrpcpp::Request& request);
+    /// Notification received from control script
+    void onControlNotification(const jsonrpcpp::Notification& notification);
+    /// Log message received from control script via stderr
+    void onControlLog(std::string line);
+    /// Send request to stream control script
+    void sendRequest(const std::string& method, const jsonrpcpp::Parameter& params, ResultHandler handler);
+
     std::chrono::time_point<std::chrono::steady_clock> tvEncodedChunk_;
     std::vector<PcmListener*> pcmListeners_;
     StreamUri uri_;
@@ -177,6 +199,7 @@ protected:
     mutable std::recursive_mutex mutex_;
 };
 
+
 } // namespace streamreader
 
 #endif

test/CMakeLists.txt

@@ -9,7 +9,11 @@ if (ANDROID)
 endif (ANDROID)
 
 # Make test executable
-set(TEST_SOURCES ${CMAKE_CURRENT_SOURCE_DIR}/test_main.cpp ${CMAKE_SOURCE_DIR}/server/streamreader/stream_uri.cpp)
+set(TEST_SOURCES 
+    ${CMAKE_CURRENT_SOURCE_DIR}/test_main.cpp 
+    ${CMAKE_SOURCE_DIR}/server/streamreader/control_error.cpp
+    ${CMAKE_SOURCE_DIR}/server/streamreader/stream_uri.cpp)
+
 add_executable(snapcast_test ${TEST_SOURCES})
 target_link_libraries(snapcast_test ${TEST_LIBRARIES})
 

test/test_main.cpp

@@ -25,6 +25,7 @@
 #include "common/metatags.hpp"
 #include "common/properties.hpp"
 #include "common/utils/string_utils.hpp"
+#include "server/streamreader/control_error.hpp"
 #include "server/streamreader/stream_uri.hpp"
 
 using namespace std;
@@ -237,3 +238,17 @@ TEST_CASE("Librespot")
     for (const auto& match : m)
         std::cerr << "Match: '" << match << "'\n";
 }
+
+
+TEST_CASE("Error")
+{
+    std::error_code ec = ControlErrc::can_not_control;
+    REQUIRE(ec);
+    REQUIRE(ec == ControlErrc::can_not_control);
+    REQUIRE(ec != ControlErrc::success);
+    std::cout << ec << std::endl;
+
+    ec = make_error_code(ControlErrc::can_not_control);
+    REQUIRE(ec.category() == snapcast::error::control::category());
+    std::cout << "Category: " << ec.category().name() << ", " << ec.message() << std::endl;
+}