Move StreamCtrl into separate file

server/CMakeLists.txt

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

server/streamreader/pcm_stream.cpp

@@ -34,154 +34,6 @@ namespace streamreader
 {
 
 static constexpr auto LOG_TAG = "PcmStream";
-static constexpr auto SCRIPT_LOG_TAG = "Script";
-
-
-CtrlScript::CtrlScript(boost::asio::io_context& ioc, const std::string& script) : ioc_(ioc), script_(script)
-{
-}
-
-
-CtrlScript::~CtrlScript()
-{
-    stop();
-}
-
-
-void CtrlScript::start(const std::string& stream_id, const ServerSettings& server_setttings, const OnNotification& notification_handler,
-                       const OnRequest& request_handler, const OnLog& log_handler)
-{
-    notification_handler_ = notification_handler;
-    request_handler_ = request_handler;
-    log_handler_ = log_handler;
-    pipe_stderr_ = bp::pipe();
-    pipe_stdout_ = bp::pipe();
-    stringstream params;
-    params << " \"--stream=" + stream_id + "\"";
-    if (server_setttings.http.enabled)
-        params << " --snapcast-port=" << server_setttings.http.port;
-    process_ = bp::child(
-        script_ + params.str(), bp::std_out > pipe_stdout_, bp::std_err > pipe_stderr_, bp::std_in < in_,
-        bp::on_exit =
-            [](int exit, const std::error_code& ec_in) {
-                auto severity = AixLog::Severity::debug;
-                if (exit != 0)
-                    severity = AixLog::Severity::error;
-                LOG(severity, SCRIPT_LOG_TAG) << "Exit code: " << exit << ", message: " << ec_in.message() << "\n";
-            },
-        ioc_);
-    stream_stdout_ = make_unique<boost::asio::posix::stream_descriptor>(ioc_, pipe_stdout_.native_source());
-    stream_stderr_ = make_unique<boost::asio::posix::stream_descriptor>(ioc_, pipe_stderr_.native_source());
-    stdoutReadLine();
-    stderrReadLine();
-}
-
-
-void CtrlScript::send(const jsonrpcpp::Request& request, const OnResponse& response_handler)
-{
-    if (response_handler)
-        request_callbacks_[request.id()] = response_handler;
-
-    std::string msg = request.to_json().dump() + "\n";
-    LOG(INFO, SCRIPT_LOG_TAG) << "Sending request: " << msg;
-    in_.write(msg.data(), msg.size());
-    in_.flush();
-}
-
-
-void CtrlScript::stderrReadLine()
-{
-    const std::string delimiter = "\n";
-    boost::asio::async_read_until(*stream_stderr_, streambuf_stderr_, delimiter, [this, delimiter](const std::error_code& ec, std::size_t bytes_transferred) {
-        if (ec)
-        {
-            LOG(ERROR, SCRIPT_LOG_TAG) << "Error while reading from stderr: " << ec.message() << "\n";
-            return;
-        }
-
-        // Extract up to the first delimiter.
-        std::string line{buffers_begin(streambuf_stderr_.data()), buffers_begin(streambuf_stderr_.data()) + bytes_transferred - delimiter.length()};
-        log_handler_(std::move(line));
-        streambuf_stderr_.consume(bytes_transferred);
-        stderrReadLine();
-    });
-}
-
-
-void CtrlScript::stdoutReadLine()
-{
-    const std::string delimiter = "\n";
-    boost::asio::async_read_until(*stream_stdout_, streambuf_stdout_, delimiter, [this, delimiter](const std::error_code& ec, std::size_t bytes_transferred) {
-        if (ec)
-        {
-            LOG(ERROR, SCRIPT_LOG_TAG) << "Error while reading from stdout: " << ec.message() << "\n";
-            return;
-        }
-
-        // Extract up to the first delimiter.
-        std::string line{buffers_begin(streambuf_stdout_.data()), buffers_begin(streambuf_stdout_.data()) + bytes_transferred - delimiter.length()};
-
-        jsonrpcpp::entity_ptr entity(nullptr);
-        try
-        {
-            entity = jsonrpcpp::Parser::do_parse(line);
-            if (!entity)
-            {
-                LOG(ERROR, LOG_TAG) << "Failed to parse message\n";
-            }
-            if (entity->is_notification())
-            {
-                jsonrpcpp::notification_ptr notification = dynamic_pointer_cast<jsonrpcpp::Notification>(entity);
-                notification_handler_(*notification);
-            }
-            else if (entity->is_request())
-            {
-                jsonrpcpp::request_ptr request = dynamic_pointer_cast<jsonrpcpp::Request>(entity);
-                request_handler_(*request);
-            }
-            else if (entity->is_response())
-            {
-                jsonrpcpp::response_ptr response = dynamic_pointer_cast<jsonrpcpp::Response>(entity);
-                LOG(INFO, LOG_TAG) << "Response: " << response->to_json() << ", id: " << response->id() << "\n";
-                auto iter = request_callbacks_.find(response->id());
-                if (iter != request_callbacks_.end())
-                {
-                    iter->second(*response);
-                    request_callbacks_.erase(iter);
-                }
-                else
-                {
-                    LOG(WARNING, LOG_TAG) << "No request found for response with id: " << response->id() << "\n";
-                }
-            }
-            else
-            {
-                LOG(WARNING, LOG_TAG) << "Not handling message: " << line << "\n";
-            }
-        }
-        catch (const jsonrpcpp::ParseErrorException& e)
-        {
-            LOG(ERROR, LOG_TAG) << "Failed to parse message: " << e.what() << "\n";
-        }
-        catch (const std::exception& e)
-        {
-            LOG(ERROR, LOG_TAG) << "Failed to parse message: " << e.what() << "\n";
-        }
-
-        streambuf_stdout_.consume(bytes_transferred);
-        stdoutReadLine();
-    });
-}
-
-
-
-void CtrlScript::stop()
-{
-    if (process_.running())
-    {
-        ::kill(-process_.native_handle(), SIGINT);
-    }
-}
 
 
 PcmStream::PcmStream(PcmListener* pcmListener, boost::asio::io_context& ioc, const ServerSettings& server_settings, const StreamUri& uri)
@@ -203,7 +55,7 @@ PcmStream::PcmStream(PcmListener* pcmListener, boost::asio::io_context& ioc, con
 
     if (uri_.query.find(kControlScript) != uri_.query.end())
     {
-        ctrl_script_ = std::make_unique<CtrlScript>(ioc, uri_.query[kControlScript]);
+        stream_ctrl_ = std::make_unique<ScriptStreamControl>(ioc, uri_.query[kControlScript]);
     }
 
     if (uri_.query.find(kUriChunkMs) != uri_.query.end())
@@ -277,12 +129,12 @@ void PcmStream::onControlNotification(const jsonrpcpp::Notification& notificatio
         else if (notification.method() == "Plugin.Stream.Ready")
         {
             LOG(DEBUG, LOG_TAG) << "Plugin is ready\n";
-            ctrl_script_->send({++req_id_, "Plugin.Stream.Player.GetProperties"}, [this](const jsonrpcpp::Response& response) {
+            stream_ctrl_->command({++req_id_, "Plugin.Stream.Player.GetProperties"}, [this](const jsonrpcpp::Response& response) {
                 LOG(INFO, LOG_TAG) << "Response for Plugin.Stream.Player.GetProperties: " << response.to_json() << "\n";
                 if (response.error().code() == 0)
                     setProperties(response.result());
             });
-            ctrl_script_->send({++req_id_, "Plugin.Stream.Player.GetMetadata"}, [this](const jsonrpcpp::Response& response) {
+            stream_ctrl_->command({++req_id_, "Plugin.Stream.Player.GetMetadata"}, [this](const jsonrpcpp::Response& response) {
                 LOG(INFO, LOG_TAG) << "Response for Plugin.Stream.Player.GetMetadata: " << response.to_json() << "\n";
                 if (response.error().code() == 0)
                     setMetadata(response.result());
@@ -324,7 +176,7 @@ void PcmStream::onControlLog(std::string line)
         severity = AixLog::Severity::error;
     else if ((tmp.find(" fatal") != string::npos) || (tmp.find(" critical") != string::npos))
         severity = AixLog::Severity::fatal;
-    LOG(severity, SCRIPT_LOG_TAG) << "Stream: " << getId() << ", message: " << line << "\n";
+    LOG(severity, LOG_TAG) << "Stream: " << getId() << ", message: " << line << "\n";
 }
 
 
@@ -336,9 +188,9 @@ void PcmStream::start()
                    sampleFormat_);
     active_ = true;
 
-    if (ctrl_script_)
+    if (stream_ctrl_)
     {
-        ctrl_script_->start(
+        stream_ctrl_->start(
             getId(), server_settings_, [this](const jsonrpcpp::Notification& notification) { onControlNotification(notification); },
             [this](const jsonrpcpp::Request& request) { onControlRequest(request); }, [this](std::string message) { onControlLog(std::move(message)); });
     }
@@ -449,7 +301,7 @@ const Properties& PcmStream::getProperties() const
 }
 
 
-void PcmStream::setProperty(const jsonrpcpp::Request& request, const CtrlScript::OnResponse& response_handler)
+void PcmStream::setProperty(const jsonrpcpp::Request& request, const StreamControl::OnResponse& response_handler)
 {
     try
     {
@@ -488,10 +340,10 @@ void PcmStream::setProperty(const jsonrpcpp::Request& request, const CtrlScript:
         if (!properties_.can_control)
             throw SnapException("CanControl is false");
 
-        if (ctrl_script_)
+        if (stream_ctrl_)
         {
             jsonrpcpp::Request req(++req_id_, "Plugin.Stream.Player.SetProperty", {name, value});
-            ctrl_script_->send(req, response_handler);
+            stream_ctrl_->command(req, response_handler);
         }
     }
     catch (const std::exception& e)
@@ -503,7 +355,7 @@ void PcmStream::setProperty(const jsonrpcpp::Request& request, const CtrlScript:
 }
 
 
-void PcmStream::control(const jsonrpcpp::Request& request, const CtrlScript::OnResponse& response_handler)
+void PcmStream::control(const jsonrpcpp::Request& request, const StreamControl::OnResponse& response_handler)
 {
     try
     {
@@ -511,27 +363,56 @@ void PcmStream::control(const jsonrpcpp::Request& request, const CtrlScript::OnR
             throw SnapException("Parameter 'command' is missing");
 
         std::string command = request.params().get("command");
-        static std::set<std::string> supported_commands{"Next", "Previous", "Pause", "PlayPause", "Stop", "Play", "Seek", "SetPosition"};
+        if (command == "SetPosition")
-        if (supported_commands.find(command) == supported_commands.end())
+        {
+            if (!request.params().has("params") || !request.params().get("params").contains("Position") || !request.params().get("params").contains("TrackId"))
+                throw SnapException("SetPosition requires parameters 'Position' and 'TrackId'");
+            if (!properties_.can_seek)
+                throw SnapException("CanSeek is false");
+        }
+        else if (command == "Seek")
+        {
+            if (!request.params().has("params") || !request.params().get("params").contains("Offset"))
+                throw SnapException("Seek requires parameter 'Offset'");
+            if (!properties_.can_seek)
+                throw SnapException("CanSeek is false");
+        }
+        else if (command == "Next")
+        {
+            if (!properties_.can_go_next)
+                throw SnapException("CanGoNext is false");
+        }
+        else if (command == "Previous")
+        {
+            if (!properties_.can_go_previous)
+                throw SnapException("CanGoPrevious is false");
+        }
+        else if ((command == "Pause") || (command == "PlayPause"))
+        {
+            if (!properties_.can_pause)
+                throw SnapException("CanPause is false");
+        }
+        else if (command == "Stop")
+        {
+            if (!properties_.can_control)
+                throw SnapException("CanControl is false");
+        }
+        else if (command == "Play")
+        {
+            if (!properties_.can_play)
+                throw SnapException("CanPlay is false");
+        }
+        else
             throw SnapException("Command not supported");
 
-        if (((command == "SetPosition") || (command == "Seek")) && !request.params().has("params"))
-            throw SnapException("Parameter 'params' is missing");
-
-        if ((command == "SetPosition") && (!request.params().get("params").contains("Position") || !request.params().get("params").contains("TrackId")))
-            throw SnapException("SetPosition requires parameters 'Position' and 'TrackId'");
-
-        if ((command == "Seek") && !request.params().get("params").contains("Offset"))
-            throw SnapException("Seek requires parameter 'Offset'");
-
         LOG(INFO, LOG_TAG) << "Stream '" << getId() << "' received command: '" << command << "', params: '" << request.params().to_json() << "'\n";
-        if (ctrl_script_)
+        if (stream_ctrl_)
         {
             jsonrpcpp::Parameter params{"command", command};
             if (request.params().has("params"))
                 params.add("params", request.params().get("params"));
             jsonrpcpp::Request req(++req_id_, "Plugin.Stream.Player.Control", params);
-            ctrl_script_->send(req, response_handler);
+            stream_ctrl_->command(req, response_handler);
         }
     }
     catch (const std::exception& e)

server/streamreader/pcm_stream.hpp

@@ -19,15 +19,7 @@
 #ifndef PCM_STREAM_HPP
 #define PCM_STREAM_HPP
 
-#pragma GCC diagnostic push
-#pragma GCC diagnostic ignored "-Wunused-result"
-#pragma GCC diagnostic ignored "-Wunused-parameter"
-#pragma GCC diagnostic ignored "-Wmissing-braces"
-#include <boost/process.hpp>
-#pragma GCC diagnostic pop
 #include <atomic>
-#include <condition_variable>
-#include <map>
 #include <string>
 #include <vector>
 
@@ -40,10 +32,10 @@
 #include "common/properties.hpp"
 #include "common/sample_format.hpp"
 #include "encoder/encoder.hpp"
-#include "message/codec_header.hpp"
-// #include "message/stream_tags.hpp"
 #include "jsonrpcpp.hpp"
+#include "message/codec_header.hpp"
 #include "server_settings.hpp"
+#include "stream_control.hpp"
 #include "stream_uri.hpp"
 
 
@@ -113,45 +105,6 @@ public:
 };
 
 
-class CtrlScript
-{
-public:
-    using OnRequest = std::function<void(const jsonrpcpp::Request& response)>;
-    using OnNotification = std::function<void(const jsonrpcpp::Notification& response)>;
-    using OnResponse = std::function<void(const jsonrpcpp::Response& response)>;
-    using OnLog = std::function<void(std::string message)>;
-
-    CtrlScript(boost::asio::io_context& ioc, const std::string& script);
-    virtual ~CtrlScript();
-
-    void start(const std::string& stream_id, const ServerSettings& server_setttings, const OnNotification& notification_handler,
-               const OnRequest& request_handler, const OnLog& log_handler);
-    void stop();
-    /// Send a message to stdin of the process
-    void send(const jsonrpcpp::Request& request, const OnResponse& response_handler);
-
-private:
-    void stderrReadLine();
-    void stdoutReadLine();
-
-    bp::child process_;
-    bp::pipe pipe_stdout_;
-    bp::pipe pipe_stderr_;
-    std::unique_ptr<boost::asio::posix::stream_descriptor> stream_stdout_;
-    std::unique_ptr<boost::asio::posix::stream_descriptor> stream_stderr_;
-    boost::asio::streambuf streambuf_stdout_;
-    boost::asio::streambuf streambuf_stderr_;
-    OnRequest request_handler_;
-    OnNotification notification_handler_;
-    OnLog log_handler_;
-
-    boost::asio::io_context& ioc_;
-    std::string script_;
-    bp::opstream in_;
-
-    std::map<jsonrpcpp::Id, OnResponse> request_callbacks_;
-};
-
 
 /// Reads and decodes PCM data
 /**
@@ -180,8 +133,8 @@ public:
     const Metatags& getMetadata() const;
     const Properties& getProperties() const;
 
-    virtual void setProperty(const jsonrpcpp::Request& request, const CtrlScript::OnResponse& response_handler);
+    virtual void setProperty(const jsonrpcpp::Request& request, const StreamControl::OnResponse& response_handler);
-    virtual void control(const jsonrpcpp::Request& request, const CtrlScript::OnResponse& response_handler);
+    virtual void control(const jsonrpcpp::Request& request, const StreamControl::OnResponse& response_handler);
 
     virtual ReaderState getState() const;
     virtual json toJson() const;
@@ -215,7 +168,7 @@ protected:
     Properties properties_;
     boost::asio::io_context& ioc_;
     ServerSettings server_settings_;
-    std::unique_ptr<CtrlScript> ctrl_script_;
+    std::unique_ptr<StreamControl> stream_ctrl_;
     int req_id_;
 };
 

server/streamreader/stream_control.cpp

@@ -0,0 +1,235 @@
+/***
+    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 <fcntl.h>
+#include <memory>
+#include <sys/stat.h>
+
+#include "common/aixlog.hpp"
+#include "common/snap_exception.hpp"
+#include "common/str_compat.hpp"
+#include "common/utils/string_utils.hpp"
+#include "encoder/encoder_factory.hpp"
+#include "stream_control.hpp"
+
+
+using namespace std;
+
+namespace streamreader
+{
+
+static constexpr auto LOG_TAG = "Script";
+
+
+StreamControl::StreamControl(boost::asio::io_context& ioc) : ioc_(ioc)
+{
+}
+
+
+StreamControl::~StreamControl()
+{
+    stop();
+}
+
+
+void StreamControl::start(const std::string& stream_id, const ServerSettings& server_setttings, const OnNotification& notification_handler,
+                          const OnRequest& request_handler, const OnLog& log_handler)
+{
+    notification_handler_ = notification_handler;
+    request_handler_ = request_handler;
+    log_handler_ = log_handler;
+
+    doStart(stream_id, server_setttings);
+}
+
+
+void StreamControl::command(const jsonrpcpp::Request& request, const OnResponse& response_handler)
+{
+    if (response_handler)
+        request_callbacks_[request.id()] = response_handler;
+
+    doCommand(request);
+}
+
+
+void StreamControl::stop()
+{
+}
+
+
+void StreamControl::onNotification(const jsonrpcpp::Notification& notification)
+{
+    notification_handler_(notification);
+}
+
+
+void StreamControl::onRequest(const jsonrpcpp::Request& request)
+{
+    request_handler_(request);
+}
+
+
+void StreamControl::onResponse(const jsonrpcpp::Response& response)
+{
+    LOG(INFO, LOG_TAG) << "Response: " << response.to_json() << ", id: " << response.id() << "\n";
+    // TODO: call request_callbacks_ on timeout with error
+    auto iter = request_callbacks_.find(response.id());
+    if (iter != request_callbacks_.end())
+    {
+        iter->second(response);
+        request_callbacks_.erase(iter);
+    }
+    else
+    {
+        LOG(WARNING, LOG_TAG) << "No request found for response with id: " << response.id() << "\n";
+    }
+}
+
+
+void StreamControl::onLog(std::string message)
+{
+    log_handler_(std::move(message));
+}
+
+
+
+ScriptStreamControl::ScriptStreamControl(boost::asio::io_context& ioc, const std::string& script) : StreamControl(ioc), script_(script)
+{
+}
+
+
+void ScriptStreamControl::doStart(const std::string& stream_id, const ServerSettings& server_setttings)
+{
+    pipe_stderr_ = bp::pipe();
+    pipe_stdout_ = bp::pipe();
+    stringstream params;
+    params << " \"--stream=" + stream_id + "\"";
+    if (server_setttings.http.enabled)
+        params << " --snapcast-port=" << server_setttings.http.port;
+    process_ = bp::child(
+        script_ + params.str(), bp::std_out > pipe_stdout_, bp::std_err > pipe_stderr_, bp::std_in < in_,
+        bp::on_exit =
+            [](int exit, const std::error_code& ec_in) {
+                auto severity = AixLog::Severity::debug;
+                if (exit != 0)
+                    severity = AixLog::Severity::error;
+                LOG(severity, LOG_TAG) << "Exit code: " << exit << ", message: " << ec_in.message() << "\n";
+            },
+        ioc_);
+    stream_stdout_ = make_unique<boost::asio::posix::stream_descriptor>(ioc_, pipe_stdout_.native_source());
+    stream_stderr_ = make_unique<boost::asio::posix::stream_descriptor>(ioc_, pipe_stderr_.native_source());
+    stdoutReadLine();
+    stderrReadLine();
+}
+
+
+void ScriptStreamControl::doCommand(const jsonrpcpp::Request& request)
+{
+    std::string msg = request.to_json().dump() + "\n";
+    LOG(INFO, LOG_TAG) << "Sending request: " << msg;
+    in_.write(msg.data(), msg.size());
+    in_.flush();
+}
+
+
+void ScriptStreamControl::stderrReadLine()
+{
+    const std::string delimiter = "\n";
+    boost::asio::async_read_until(*stream_stderr_, streambuf_stderr_, delimiter, [this, delimiter](const std::error_code& ec, std::size_t bytes_transferred) {
+        if (ec)
+        {
+            LOG(ERROR, LOG_TAG) << "Error while reading from stderr: " << ec.message() << "\n";
+            return;
+        }
+
+        // Extract up to the first delimiter.
+        std::string line{buffers_begin(streambuf_stderr_.data()), buffers_begin(streambuf_stderr_.data()) + bytes_transferred - delimiter.length()};
+        onLog(std::move(line));
+        streambuf_stderr_.consume(bytes_transferred);
+        stderrReadLine();
+    });
+}
+
+
+void ScriptStreamControl::stdoutReadLine()
+{
+    const std::string delimiter = "\n";
+    boost::asio::async_read_until(*stream_stdout_, streambuf_stdout_, delimiter, [this, delimiter](const std::error_code& ec, std::size_t bytes_transferred) {
+        if (ec)
+        {
+            LOG(ERROR, LOG_TAG) << "Error while reading from stdout: " << ec.message() << "\n";
+            return;
+        }
+
+        // Extract up to the first delimiter.
+        std::string line{buffers_begin(streambuf_stdout_.data()), buffers_begin(streambuf_stdout_.data()) + bytes_transferred - delimiter.length()};
+
+        jsonrpcpp::entity_ptr entity(nullptr);
+        try
+        {
+            entity = jsonrpcpp::Parser::do_parse(line);
+            if (!entity)
+            {
+                LOG(ERROR, LOG_TAG) << "Failed to parse message\n";
+            }
+            if (entity->is_notification())
+            {
+                jsonrpcpp::notification_ptr notification = dynamic_pointer_cast<jsonrpcpp::Notification>(entity);
+                onNotification(*notification);
+            }
+            else if (entity->is_request())
+            {
+                jsonrpcpp::request_ptr request = dynamic_pointer_cast<jsonrpcpp::Request>(entity);
+                onRequest(*request);
+            }
+            else if (entity->is_response())
+            {
+                jsonrpcpp::response_ptr response = dynamic_pointer_cast<jsonrpcpp::Response>(entity);
+                onResponse(*response);
+            }
+            else
+            {
+                LOG(WARNING, LOG_TAG) << "Not handling message: " << line << "\n";
+            }
+        }
+        catch (const jsonrpcpp::ParseErrorException& e)
+        {
+            LOG(ERROR, LOG_TAG) << "Failed to parse message: " << e.what() << "\n";
+        }
+        catch (const std::exception& e)
+        {
+            LOG(ERROR, LOG_TAG) << "Failed to parse message: " << e.what() << "\n";
+        }
+
+        streambuf_stdout_.consume(bytes_transferred);
+        stdoutReadLine();
+    });
+}
+
+
+
+void ScriptStreamControl::stop()
+{
+    if (process_.running())
+    {
+        ::kill(-process_.native_handle(), SIGINT);
+    }
+}
+
+
+} // namespace streamreader

server/streamreader/stream_control.hpp

@@ -0,0 +1,112 @@
+/***
+    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 STREAM_CONTROL_HPP
+#define STREAM_CONTROL_HPP
+
+#pragma GCC diagnostic push
+#pragma GCC diagnostic ignored "-Wunused-result"
+#pragma GCC diagnostic ignored "-Wunused-parameter"
+#pragma GCC diagnostic ignored "-Wmissing-braces"
+#include <boost/process.hpp>
+#pragma GCC diagnostic pop
+#include <map>
+#include <string>
+
+#include <boost/asio/io_context.hpp>
+#include <boost/asio/read_until.hpp>
+
+#include "jsonrpcpp.hpp"
+#include "server_settings.hpp"
+
+
+namespace bp = boost::process;
+using json = nlohmann::json;
+
+
+namespace streamreader
+{
+
+
+class StreamControl
+{
+public:
+    using OnRequest = std::function<void(const jsonrpcpp::Request& response)>;
+    using OnNotification = std::function<void(const jsonrpcpp::Notification& response)>;
+    using OnResponse = std::function<void(const jsonrpcpp::Response& response)>;
+    using OnLog = std::function<void(std::string message)>;
+
+    StreamControl(boost::asio::io_context& ioc);
+    virtual ~StreamControl();
+
+    void start(const std::string& stream_id, const ServerSettings& server_setttings, const OnNotification& notification_handler,
+               const OnRequest& request_handler, const OnLog& log_handler);
+    virtual void stop();
+
+    void command(const jsonrpcpp::Request& request, const OnResponse& response_handler);
+
+protected:
+    virtual void doCommand(const jsonrpcpp::Request& request) = 0;
+    virtual void doStart(const std::string& stream_id, const ServerSettings& server_setttings) = 0;
+
+    void onNotification(const jsonrpcpp::Notification& notification);
+    void onRequest(const jsonrpcpp::Request& request);
+    void onResponse(const jsonrpcpp::Response& response);
+    void onLog(std::string message);
+
+    OnRequest request_handler_;
+    OnNotification notification_handler_;
+    OnLog log_handler_;
+
+    boost::asio::io_context& ioc_;
+    std::map<jsonrpcpp::Id, OnResponse> request_callbacks_;
+};
+
+
+class ScriptStreamControl : public StreamControl
+{
+public:
+    ScriptStreamControl(boost::asio::io_context& ioc, const std::string& script);
+    virtual ~ScriptStreamControl() = default;
+
+    void stop() override;
+
+protected:
+    /// Send a message to stdin of the process
+    void doCommand(const jsonrpcpp::Request& request) override;
+    void doStart(const std::string& stream_id, const ServerSettings& server_setttings) override;
+
+    void stderrReadLine();
+    void stdoutReadLine();
+
+    bp::child process_;
+    bp::pipe pipe_stdout_;
+    bp::pipe pipe_stderr_;
+    std::unique_ptr<boost::asio::posix::stream_descriptor> stream_stdout_;
+    std::unique_ptr<boost::asio::posix::stream_descriptor> stream_stderr_;
+    boost::asio::streambuf streambuf_stdout_;
+    boost::asio::streambuf streambuf_stderr_;
+
+    std::string script_;
+    bp::opstream in_;
+};
+
+
+} // namespace streamreader
+
+#endif