add HTTP and websocket support (preliminary)

server/CMakeLists.txt

@@ -2,7 +2,7 @@ set(SERVER_SOURCES
     config.cpp
     controlServer.cpp
     control_session_tcp.cpp
-    control_session_ws.cpp
+    control_session_http.cpp
     snapServer.cpp
     streamServer.cpp
     streamSession.cpp

server/controlServer.cpp

@@ -21,8 +21,8 @@
 #include "common/snapException.h"
 #include "common/utils.h"
 #include "config.h"
+#include "control_session_http.hpp"
 #include "control_session_tcp.hpp"
-#include "control_session_ws.hpp"
 #include "jsonrpcpp.hpp"
 #include "message/time.h"
 
@@ -74,12 +74,13 @@ void ControlServer::send(const std::string& message, const ControlSession* exclu
 }
 
 
-void ControlServer::onMessageReceived(ControlSession* connection, const std::string& message)
+std::string ControlServer::onMessageReceived(ControlSession* connection, const std::string& message)
 {
     std::lock_guard<std::recursive_mutex> mlock(session_mutex_);
     LOG(DEBUG) << "received: \"" << message << "\"\n";
     if (controlMessageReceiver_ != nullptr)
-        controlMessageReceiver_->onMessageReceived(connection, message);
+        return controlMessageReceiver_->onMessageReceived(connection, message);
+    return "";
 }
 
 
@@ -92,9 +93,9 @@ void ControlServer::startAccept()
             LOG(ERROR) << "Error while accepting socket connection: " << ec.message() << "\n";
     };
 
-    auto accept_handler_ws = [this](error_code ec, tcp::socket socket) {
+    auto accept_handler_http = [this](error_code ec, tcp::socket socket) {
         if (!ec)
-            handleAccept<ControlSessionWs>(std::move(socket));
+            handleAccept<ControlSessionHttp>(std::move(socket));
         else
             LOG(ERROR) << "Error while accepting socket connection: " << ec.message() << "\n";
     };
@@ -105,11 +106,11 @@ void ControlServer::startAccept()
     if (acceptor_tcp_.second)
         acceptor_tcp_.second->async_accept(accept_handler_tcp);
 
-    if (acceptor_ws_.first)
-        acceptor_ws_.first->async_accept(accept_handler_ws);
+    if (acceptor_http_.first)
+        acceptor_http_.first->async_accept(accept_handler_http);
 
-    if (acceptor_ws_.second)
-        acceptor_ws_.second->async_accept(accept_handler_ws);
+    if (acceptor_http_.second)
+        acceptor_http_.second->async_accept(accept_handler_http);
 }
 
 
@@ -179,7 +180,7 @@ std::pair<acceptor_ptr, acceptor_ptr> ControlServer::createAcceptors(size_t port
 void ControlServer::start()
 {
     acceptor_tcp_ = createAcceptors(port_);
-    acceptor_ws_ = createAcceptors(8080);
+    acceptor_http_ = createAcceptors(8080);
     startAccept();
 }
 
@@ -196,8 +197,8 @@ void ControlServer::stop()
 
     cancel_accept(acceptor_tcp_.first.get());
     cancel_accept(acceptor_tcp_.second.get());
-    cancel_accept(acceptor_ws_.first.get());
-    cancel_accept(acceptor_ws_.second.get());
+    cancel_accept(acceptor_http_.first.get());
+    cancel_accept(acceptor_http_.second.get());
     std::lock_guard<std::recursive_mutex> mlock(session_mutex_);
     for (auto s : sessions_)
     {

server/controlServer.h

@@ -54,7 +54,7 @@ public:
     void send(const std::string& message, const ControlSession* excludeSession = nullptr);
 
     /// Clients call this when they receive a message. Implementation of MessageReceiver::onMessageReceived
-    void onMessageReceived(ControlSession* connection, const std::string& message) override;
+    std::string onMessageReceived(ControlSession* connection, const std::string& message) override;
 
 private:
     void startAccept();
@@ -69,7 +69,7 @@ private:
     std::vector<std::weak_ptr<ControlSession>> sessions_;
 
     std::pair<acceptor_ptr, acceptor_ptr> acceptor_tcp_;
-    std::pair<acceptor_ptr, acceptor_ptr> acceptor_ws_;
+    std::pair<acceptor_ptr, acceptor_ptr> acceptor_http_;
 
     boost::asio::io_context* io_context_;
     size_t port_;

server/control_session.hpp

@@ -41,7 +41,8 @@ class ControlSession;
 class ControlMessageReceiver
 {
 public:
-    virtual void onMessageReceived(ControlSession* connection, const std::string& message) = 0;
+    // TODO: rename, error handling
+    virtual std::string onMessageReceived(ControlSession* connection, const std::string& message) = 0;
 };
 
 

server/control_session_http.cpp

@@ -0,0 +1,268 @@
+/***
+    This file is part of snapcast
+    Copyright (C) 2014-2019  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_session_http.hpp"
+#include "aixlog.hpp"
+#include "message/pcmChunk.h"
+#include <boost/beast/http/file_body.hpp>
+#include <iostream>
+
+using namespace std;
+
+
+ControlSessionHttp::ControlSessionHttp(ControlMessageReceiver* receiver, tcp::socket&& socket) : ControlSession(receiver), socket_(std::move(socket))
+{
+    LOG(DEBUG) << "ControlSessionHttp\n";
+}
+
+
+ControlSessionHttp::~ControlSessionHttp()
+{
+    LOG(DEBUG) << "ControlSessionHttp::~ControlSessionHttp()\n";
+    stop();
+}
+
+
+void ControlSessionHttp::start()
+{
+    auto self = shared_from_this();
+    http::async_read(socket_, buffer_, req_, [this, self](boost::system::error_code ec, std::size_t bytes) { on_read(ec, bytes); });
+}
+
+
+// This function produces an HTTP response for the given
+// request. The type of the response object depends on the
+// contents of the request, so the interface requires the
+// caller to pass a generic lambda for receiving the response.
+template <class Body, class Allocator, class Send>
+void ControlSessionHttp::handle_request(http::request<Body, http::basic_fields<Allocator>>&& req, Send&& send)
+{
+    // Returns a bad request response
+    auto const bad_request = [&req](boost::beast::string_view why) {
+        http::response<http::string_body> res{http::status::bad_request, req.version()};
+        res.set(http::field::server, BOOST_BEAST_VERSION_STRING);
+        res.set(http::field::content_type, "text/html");
+        res.keep_alive(req.keep_alive());
+        res.body() = why.to_string();
+        res.prepare_payload();
+        return res;
+    };
+
+    // Returns a not found response
+    auto const not_found = [&req](boost::beast::string_view target) {
+        http::response<http::string_body> res{http::status::not_found, req.version()};
+        res.set(http::field::server, BOOST_BEAST_VERSION_STRING);
+        res.set(http::field::content_type, "text/html");
+        res.keep_alive(req.keep_alive());
+        res.body() = "The resource '" + target.to_string() + "' was not found.";
+        res.prepare_payload();
+        return res;
+    };
+
+    // Returns a server error response
+    auto const server_error = [&req](boost::beast::string_view what) {
+        http::response<http::string_body> res{http::status::internal_server_error, req.version()};
+        res.set(http::field::server, BOOST_BEAST_VERSION_STRING);
+        res.set(http::field::content_type, "text/html");
+        res.keep_alive(req.keep_alive());
+        res.body() = "An error occurred: '" + what.to_string() + "'";
+        res.prepare_payload();
+        return res;
+    };
+
+    // Make sure we can handle the method
+    if (req.method() != http::verb::post)
+        return send(bad_request("Unknown HTTP-method"));
+
+    // Request path must be absolute and not contain "..".
+    // if (req.target().empty() || req.target()[0] != '/' || req.target().find("..") != boost::beast::string_view::npos)
+    //     return send(bad_request("Illegal request-target"));
+
+    LOG(DEBUG) << "content type: " << req[beast::http::field::content_type] << "\n";
+    LOG(DEBUG) << "body: " << req.body() << "\n";
+
+    // TODO: error handling: bad request, ...
+    string response = message_receiver_->onMessageReceived(this, req.body());
+
+    http::response<http::string_body> res{http::status::ok, req.version()};
+    res.set(http::field::server, BOOST_BEAST_VERSION_STRING);
+    res.set(http::field::content_type, "application/json");
+    res.keep_alive(req.keep_alive());
+    res.body() = response; // R"({"jsonrpc": "2.0", "id": 1, "result": "stopped"})";
+    res.prepare_payload();
+    return send(std::move(res));
+}
+
+void ControlSessionHttp::on_read(beast::error_code ec, std::size_t bytes_transferred)
+{
+    // This means they closed the connection
+    if (ec == http::error::end_of_stream)
+    {
+        socket_.shutdown(tcp::socket::shutdown_send, ec);
+        return;
+    }
+
+    // Handle the error, if any
+    if (ec)
+    {
+        LOG(ERROR) << "ControlSessionHttp::on_read error: " << ec.message() << "\n";
+        return;
+    }
+
+    // TODO: error handling
+    // urls should be:
+    //  http://<host>/snapcast/rpc
+    //  ws://<host>/snapcast/ws or ws://<host>/snapcast/rpc?
+
+    // See if it is a WebSocket Upgrade
+    if (websocket::is_upgrade(req_))
+    {
+        // Create a WebSocket session by transferring the socket
+        // std::make_shared<websocket_session>(std::move(socket_), state_)->run(std::move(req_));
+        ws_ = make_unique<websocket::stream<beast::tcp_stream>>(std::move(socket_));
+        auto self = shared_from_this();
+        ws_->async_accept(req_, [this, self](beast::error_code ec) { on_accept_ws(ec); });
+        LOG(DEBUG) << "websocket upgrade\n";
+        return;
+    }
+
+    // Send the response
+    handle_request(std::move(req_), [this](auto&& response) {
+        // The lifetime of the message has to extend
+        // for the duration of the async operation so
+        // we use a shared_ptr to manage it.
+        using response_type = typename std::decay<decltype(response)>::type;
+        auto sp = std::make_shared<response_type>(std::forward<decltype(response)>(response));
+
+        // Write the response
+        auto self = this->shared_from_this();
+        http::async_write(this->socket_, *sp, [this, self, sp](beast::error_code ec, std::size_t bytes) { this->on_write(ec, bytes, sp->need_eof()); });
+    });
+}
+
+
+void ControlSessionHttp::on_write(beast::error_code ec, std::size_t, bool close)
+{
+    // Handle the error, if any
+    if (ec)
+    {
+        LOG(ERROR) << "ControlSessionHttp::on_write, error: " << ec.message() << "\n";
+        return;
+    }
+
+    if (close)
+    {
+        // This means we should close the connection, usually because
+        // the response indicated the "Connection: close" semantic.
+        socket_.shutdown(tcp::socket::shutdown_send, ec);
+        return;
+    }
+
+    // Clear contents of the request message,
+    // otherwise the read behavior is undefined.
+    req_ = {};
+
+    // Read another request
+    http::async_read(socket_, buffer_, req_, [ this, self = shared_from_this() ](beast::error_code ec, std::size_t bytes) { on_read(ec, bytes); });
+}
+
+
+void ControlSessionHttp::stop()
+{
+}
+
+
+void ControlSessionHttp::sendAsync(const std::string& message)
+{
+    if (!ws_)
+        return;
+
+    auto self(shared_from_this());
+    ws_->async_write(boost::asio::buffer(message), [this, self](std::error_code ec, std::size_t length) {
+        if (ec)
+        {
+            LOG(ERROR) << "Error while writing to control socket: " << ec.message() << "\n";
+        }
+        else
+        {
+            LOG(DEBUG) << "Wrote " << length << " bytes to control socket\n";
+        }
+    });
+}
+
+
+bool ControlSessionHttp::send(const std::string& message)
+{
+    if (!ws_)
+        return false;
+
+    boost::system::error_code ec;
+    ws_->write(boost::asio::buffer(message), ec);
+    return !ec;
+}
+
+void ControlSessionHttp::on_accept_ws(beast::error_code ec)
+{
+    if (ec)
+    {
+        LOG(ERROR) << "ControlSessionWs::on_accept, error: " << ec.message() << "\n";
+        return;
+    }
+
+    // Read a message
+    do_read_ws();
+}
+
+void ControlSessionHttp::do_read_ws()
+{
+    // Read a message into our buffer
+    auto self(shared_from_this());
+    ws_->async_read(buffer_, [this, self](beast::error_code ec, std::size_t bytes_transferred) { on_read_ws(ec, bytes_transferred); });
+}
+
+
+void ControlSessionHttp::on_read_ws(beast::error_code ec, std::size_t bytes_transferred)
+{
+    boost::ignore_unused(bytes_transferred);
+
+    // This indicates that the session was closed
+    if (ec == websocket::error::closed)
+        return;
+
+    if (ec)
+    {
+        LOG(ERROR) << "ControlSessionHttp::on_read_ws error: " << ec.message() << "\n";
+        return;
+    }
+
+    std::string line{boost::beast::buffers_to_string(buffer_.data())};
+    if (!line.empty())
+    {
+        LOG(INFO) << "received: " << line << "\n";
+        if ((message_receiver_ != nullptr) && !line.empty())
+        {
+            string response = message_receiver_->onMessageReceived(this, line);
+            if (!response.empty())
+            {
+                sendAsync(response);
+            }
+        }
+    }
+    buffer_.consume(bytes_transferred);
+    do_read_ws();
+}

server/control_session_http.hpp

@@ -16,8 +16,8 @@
     along with this program.  If not, see <http://www.gnu.org/licenses/>.
 ***/
 
-#ifndef CONTROL_SESSION_WS_HPP
-#define CONTROL_SESSION_WS_HPP
+#ifndef CONTROL_SESSION_HTTP_HPP
+#define CONTROL_SESSION_HTTP_HPP
 
 #include "control_session.hpp"
 #include <boost/beast/core.hpp>
@@ -35,12 +35,12 @@ namespace net = boost::asio;            // from <boost/asio.hpp>
  * Messages are sent to the client with the "send" method.
  * Received messages from the client are passed to the ControlMessageReceiver callback
  */
-class ControlSessionWs : public ControlSession, public std::enable_shared_from_this<ControlSession>
+class ControlSessionHttp : public ControlSession, public std::enable_shared_from_this<ControlSession>
 {
 public:
     /// ctor. Received message from the client are passed to MessageReceiver
-    ControlSessionWs(ControlMessageReceiver* receiver, tcp::socket&& socket);
-    ~ControlSessionWs() override;
+    ControlSessionHttp(ControlMessageReceiver* receiver, tcp::socket&& socket);
+    ~ControlSessionHttp() override;
     void start() override;
     void stop() override;
 
@@ -51,13 +51,26 @@ public:
     void sendAsync(const std::string& message) override;
 
 protected:
-    void on_accept(beast::error_code ec);
-    void do_read();
+    // HTTP methods
     void on_read(beast::error_code ec, std::size_t bytes_transferred);
+    void on_write(beast::error_code ec, std::size_t, bool close);
 
-    websocket::stream<beast::tcp_stream> ws_;
+    template <class Body, class Allocator, class Send>
+    void handle_request(http::request<Body, http::basic_fields<Allocator>>&& req, Send&& send);
+
+    http::request<http::string_body> req_;
+
+protected:
+    // Websocket methods
+    void on_accept_ws(beast::error_code ec);
+    void on_read_ws(beast::error_code ec, std::size_t bytes_transferred);
+    void do_read_ws();
+
+    std::unique_ptr<websocket::stream<beast::tcp_stream>> ws_;
+
+protected:
+    tcp::socket socket_;
     beast::flat_buffer buffer_;
-    beast::multi_buffer b;
 };
 
 

server/control_session_tcp.cpp

@@ -55,13 +55,18 @@ void ControlSessionTcp::do_read()
                 line.resize(line.size() - 1);
             LOG(INFO) << "received: " << line << "\n";
             if ((message_receiver_ != nullptr) && !line.empty())
-                message_receiver_->onMessageReceived(this, line);
+            {
+                string response = message_receiver_->onMessageReceived(this, line);
+                if (!response.empty())
+                    sendAsync(response);
+            }
         }
         streambuf_.consume(bytes_transferred);
         do_read();
     });
 }
 
+
 void ControlSessionTcp::start()
 {
     do_read();

server/control_session_ws.cpp

@@ -1,129 +0,0 @@
-/***
-    This file is part of snapcast
-    Copyright (C) 2014-2019  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_session_ws.hpp"
-#include "aixlog.hpp"
-#include "message/pcmChunk.h"
-#include <iostream>
-
-using namespace std;
-
-
-ControlSessionWs::ControlSessionWs(ControlMessageReceiver* receiver, tcp::socket&& socket) : ControlSession(receiver), ws_(std::move(socket))
-{
-}
-
-
-ControlSessionWs::~ControlSessionWs()
-{
-    LOG(DEBUG) << "ControlSessionWs::~ControlSessionWs()\n";
-    stop();
-}
-
-
-void ControlSessionWs::start()
-{
-    // Set suggested timeout settings for the websocket
-    ws_.set_option(websocket::stream_base::timeout::suggested(beast::role_type::server));
-
-    // Set a decorator to change the Server of the handshake
-    ws_.set_option(websocket::stream_base::decorator([](websocket::response_type& res) { res.set(http::field::server, "Snapcast websocket-server"); }));
-
-    // Accept the websocket handshake
-    auto self(shared_from_this());
-    ws_.async_accept([this, self](beast::error_code ec) { on_accept(ec); });
-}
-
-
-void ControlSessionWs::on_accept(beast::error_code ec)
-{
-    if (ec)
-    {
-        LOG(ERROR) << "ControlSessionWs::on_accept, error: " << ec.message() << "\n";
-        return;
-    }
-
-    // Read a message
-    do_read();
-}
-
-
-void ControlSessionWs::do_read()
-{
-    // Read a message into our buffer
-    auto self(shared_from_this());
-    ws_.async_read(buffer_, [this, self](beast::error_code ec, std::size_t bytes_transferred) { on_read(ec, bytes_transferred); });
-}
-
-
-void ControlSessionWs::on_read(beast::error_code ec, std::size_t bytes_transferred)
-{
-    boost::ignore_unused(bytes_transferred);
-
-    // This indicates that the session was closed
-    if (ec == websocket::error::closed)
-    {
-        LOG(INFO) << "ControlSessionWs session closed\n";
-        return;
-    }
-
-    if (ec)
-    {
-        LOG(ERROR) << "ControlSessionWs::on_read error: " << ec.message() << "\n";
-        return;
-    }
-
-    std::string line{boost::beast::buffers_to_string(buffer_.data())};
-    if (!line.empty())
-    {
-        LOG(INFO) << "received: " << line << "\n";
-        if ((message_receiver_ != nullptr) && !line.empty())
-            message_receiver_->onMessageReceived(this, line);
-    }
-    buffer_.consume(bytes_transferred);
-    do_read();
-}
-
-
-void ControlSessionWs::stop()
-{
-}
-
-
-void ControlSessionWs::sendAsync(const std::string& message)
-{
-    auto self(shared_from_this());
-    ws_.async_write(boost::asio::buffer(message), [this, self](std::error_code ec, std::size_t length) {
-        if (ec)
-        {
-            LOG(ERROR) << "Error while writing to control socket: " << ec.message() << "\n";
-        }
-        else
-        {
-            LOG(DEBUG) << "Wrote " << length << " bytes to control socket\n";
-        }
-    });
-}
-
-
-bool ControlSessionWs::send(const std::string& message)
-{
-    boost::system::error_code ec;
-    ws_.write(boost::asio::buffer(message), ec);
-    return !ec;
-}

server/streamServer.cpp

@@ -507,7 +507,7 @@ void StreamServer::ProcessRequest(const jsonrpcpp::request_ptr request, jsonrpcp
 }
 
 
-void StreamServer::onMessageReceived(ControlSession* controlSession, const std::string& message)
+std::string StreamServer::onMessageReceived(ControlSession* controlSession, const std::string& message)
 {
     LOG(DEBUG) << "onMessageReceived: " << message << "\n";
     jsonrpcpp::entity_ptr entity(nullptr);
@@ -515,17 +515,15 @@ void StreamServer::onMessageReceived(ControlSession* controlSession, const std::
     {
         entity = jsonrpcpp::Parser::do_parse(message);
         if (!entity)
-            return;
+            return "";
     }
     catch (const jsonrpcpp::ParseErrorException& e)
     {
-        controlSession->send(e.to_json().dump());
-        return;
+        return e.to_json().dump();
     }
     catch (const std::exception& e)
     {
-        controlSession->send(jsonrpcpp::ParseErrorException(e.what()).to_json().dump());
-        return;
+        return jsonrpcpp::ParseErrorException(e.what()).to_json().dump();
     }
 
     jsonrpcpp::entity_ptr response(nullptr);
@@ -535,16 +533,17 @@ void StreamServer::onMessageReceived(ControlSession* controlSession, const std::
         jsonrpcpp::request_ptr request = dynamic_pointer_cast<jsonrpcpp::Request>(entity);
         ProcessRequest(request, response, notification);
         ////cout << "Request:      " << request->to_json().dump() << "\n";
-        if (response)
-        {
-            ////cout << "Response:     " << response->to_json().dump() << "\n";
-            controlSession->send(response->to_json().dump());
-        }
         if (notification)
         {
             ////cout << "Notification: " << notification->to_json().dump() << "\n";
             controlServer_->send(notification->to_json().dump(), controlSession);
         }
+        if (response)
+        {
+            ////cout << "Response:     " << response->to_json().dump() << "\n";
+            return response->to_json().dump();
+        }
+        return "";
     }
     else if (entity->is_batch())
     {
@@ -564,11 +563,13 @@ void StreamServer::onMessageReceived(ControlSession* controlSession, const std::
                     notificationBatch.add_ptr(notification);
             }
         }
-        if (!responseBatch.entities.empty())
-            controlSession->send(responseBatch.to_json().dump());
         if (!notificationBatch.entities.empty())
             controlServer_->send(notificationBatch.to_json().dump(), controlSession);
+        if (!responseBatch.entities.empty())
+            return responseBatch.to_json().dump();
+        return "";
     }
+    return "";
 }
 
 

server/streamServer.h

@@ -83,7 +83,7 @@ public:
     void onDisconnect(StreamSession* connection) override;
 
     /// Implementation of ControllMessageReceiver::onMessageReceived, called by ControlServer::onMessageReceived
-    void onMessageReceived(ControlSession* connection, const std::string& message) override;
+    std::string onMessageReceived(ControlSession* connection, const std::string& message) override;
 
     /// Implementation of PcmListener
     void onMetaChanged(const PcmStream* pcmStream) override;