3pp-mirror/snapcast
1
0
Fork 0
mirror of https://github.com/badaix/snapcast.git synced 2025-05-25 15:06:21 +02:00
Code Issues Projects Releases Packages Wiki Activity

switch to boost::asio

Browse source
This commit is contained in:
badaix 2019-09-28 21:28:54 +02:00
parent 08d0ce58f2
commit ec3f8d8ad5
16 changed files with 70 additions and 56 deletions
Download patch file Download diff file Expand all files Collapse all files

2
CMakeLists.txt
View file

@ -199,6 +199,8 @@ if(BUILD_WITH_VORBIS)
endif(VORBISENC_FOUND)
endif()
# TODO: if (BUILD_WITH_WEBSOCKETS)
find_package(Boost 1.66 REQUIRED)
add_subdirectory(common)

12
client/clientConnection.cpp
View file

@ -48,7 +48,7 @@ void ClientConnection::socketRead(void* _to, size_t _bytes)
size_t len = 0;
do
{
len += socket_->read_some(asio::buffer((char*)_to + len, toRead));
len += socket_->read_some(boost::asio::buffer((char*)_to + len, toRead));
// cout << "len: " << len << ", error: " << error << endl;
toRead = _bytes - len;
} while (toRead > 0);
@ -71,7 +71,7 @@ std::string ClientConnection::getMacAddress() const
void ClientConnection::start()
{
tcp::resolver resolver(io_context_);
tcp::resolver::query query(host_, cpt::to_string(port_), asio::ip::resolver_query_base::numeric_service);
tcp::resolver::query query(host_, cpt::to_string(port_), boost::asio::ip::resolver_query_base::numeric_service);
auto iterator = resolver.resolve(query);
LOG(DEBUG) << "Connecting\n";
socket_.reset(new tcp::socket(io_context_));
@ -96,10 +96,10 @@ void ClientConnection::stop()
active_ = false;
try
{
std::error_code ec;
boost::system::error_code ec;
if (socket_)
{
socket_->shutdown(asio::ip::tcp::socket::shutdown_both, ec);
socket_->shutdown(boost::asio::ip::tcp::socket::shutdown_both, ec);
if (ec)
LOG(ERROR) << "Error in socket shutdown: " << ec.message() << endl;
socket_->close(ec);
@ -130,12 +130,12 @@ bool ClientConnection::send(const msg::BaseMessage* message) const
if (!connected())
return false;
// LOG(DEBUG) << "send: " << message->type << ", size: " << message->getSize() << "\n";
asio::streambuf streambuf;
boost::asio::streambuf streambuf;
std::ostream stream(&streambuf);
tv t;
message->sent = t;
message->serialize(stream);
asio::write(*socket_.get(), streambuf);
boost::asio::write(*socket_.get(), streambuf);
return true;
}

6
client/clientConnection.h
View file

@ -21,7 +21,7 @@
#include "common/timeDefs.h"
#include "message/message.h"
#include <asio.hpp>
#include <boost/asio.hpp>
#include <atomic>
#include <condition_variable>
#include <memory>
@ -31,7 +31,7 @@
#include <thread>
using asio::ip::tcp;
using boost::asio::ip::tcp;
class ClientConnection;
@ -112,7 +112,7 @@ protected:
void socketRead(void* to, size_t bytes);
void getNextMessage();
asio::io_context io_context_;
boost::asio::io_context io_context_;
mutable std::mutex socketMutex_;
std::shared_ptr<tcp::socket> socket_;
std::atomic<bool> active_;

3
server/CMakeLists.txt
View file

@ -58,6 +58,9 @@ if (FLAC_FOUND)
list(APPEND SERVER_INCLUDE ${FLAC_INCLUDE_DIRS})
endif (FLAC_FOUND)
# TODO: if (BUILD_WITH_WEBSOCKETS)
list(APPEND SERVER_LIBRARIES Boost::boost)
include_directories(${SERVER_INCLUDE})
add_executable(snapserver ${SERVER_SOURCES})
target_link_libraries(snapserver ${SERVER_LIBRARIES})

12
server/controlServer.cpp
View file

@ -30,7 +30,7 @@ using namespace std;
using json = nlohmann::json;
ControlServer::ControlServer(asio::io_context* io_context, size_t port, ControlMessageReceiver* controlMessageReceiver)
ControlServer::ControlServer(boost::asio::io_context* io_context, size_t port, ControlMessageReceiver* controlMessageReceiver)
: acceptor_v4_(nullptr), acceptor_v6_(nullptr), io_context_(io_context), port_(port), controlMessageReceiver_(controlMessageReceiver)
{
}
@ -137,14 +137,14 @@ void ControlServer::start()
try
{
acceptor_v6_ = make_shared<tcp::acceptor>(*io_context_, endpoint_v6);
error_code ec;
acceptor_v6_->set_option(asio::ip::v6_only(false), ec);
asio::ip::v6_only option;
boost::system::error_code ec;
acceptor_v6_->set_option(boost::asio::ip::v6_only(false), ec);
boost::asio::ip::v6_only option;
acceptor_v6_->get_option(option);
is_v6_only = option.value();
LOG(DEBUG) << "IPv6 only: " << is_v6_only << "\n";
}
catch (const asio::system_error& e)
catch (const boost::system::system_error& e)
{
LOG(ERROR) << "error creating TCP acceptor: " << e.what() << ", code: " << e.code() << "\n";
}
@ -156,7 +156,7 @@ void ControlServer::start()
{
acceptor_v4_ = make_shared<tcp::acceptor>(*io_context_, endpoint_v4);
}
catch (const asio::system_error& e)
catch (const boost::system::system_error& e)
{
LOG(ERROR) << "error creating TCP acceptor: " << e.what() << ", code: " << e.code() << "\n";
}

8
server/controlServer.h
View file

@ -19,7 +19,7 @@
#ifndef CONTROL_SERVER_H
#define CONTROL_SERVER_H
#include <asio.hpp>
#include <boost/asio.hpp>
#include <memory>
#include <mutex>
#include <set>
@ -34,7 +34,7 @@
#include "message/message.h"
#include "message/serverSettings.h"
using asio::ip::tcp;
using boost::asio::ip::tcp;
/// Telnet like remote control
/**
@ -43,7 +43,7 @@ using asio::ip::tcp;
class ControlServer : public ControlMessageReceiver
{
public:
ControlServer(asio::io_context* io_context, size_t port, ControlMessageReceiver* controlMessageReceiver = nullptr);
ControlServer(boost::asio::io_context* io_context, size_t port, ControlMessageReceiver* controlMessageReceiver = nullptr);
virtual ~ControlServer();
void start();
@ -65,7 +65,7 @@ private:
std::shared_ptr<tcp::acceptor> acceptor_v4_;
std::shared_ptr<tcp::acceptor> acceptor_v6_;
asio::io_context* io_context_;
boost::asio::io_context* io_context_;
size_t port_;
ControlMessageReceiver* controlMessageReceiver_;
};

7
server/control_session.hpp
View file

@ -21,7 +21,7 @@
#include "common/queue.h"
#include "message/message.h"
#include <asio.hpp>
#include <boost/asio.hpp>
#include <atomic>
#include <condition_variable>
#include <memory>
@ -31,7 +31,7 @@
#include <thread>
using asio::ip::tcp;
using boost::asio::ip::tcp;
class ControlSession;
@ -55,7 +55,7 @@ class ControlSession : public std::enable_shared_from_this<ControlSession>
{
public:
/// ctor. Received message from the client are passed to MessageReceiver
ControlSession(ControlMessageReceiver* receiver, tcp::socket&& socket) : message_receiver_(receiver), socket_(std::move(socket))
ControlSession(ControlMessageReceiver* receiver) : message_receiver_(receiver)
{
}
virtual ~ControlSession() = default;
@ -69,7 +69,6 @@ public:
virtual void sendAsync(const std::string& message) = 0;
protected:
tcp::socket socket_;
ControlMessageReceiver* message_receiver_;
};

16
server/control_session_tcp.cpp
View file

@ -19,14 +19,12 @@
#include "control_session_tcp.hpp"
#include "aixlog.hpp"
#include "message/pcmChunk.h"
#include <iostream>
#include <mutex>
using namespace std;
ControlSessionTcp::ControlSessionTcp(ControlMessageReceiver* receiver, tcp::socket&& socket) : ControlSession(receiver, std::move(socket))
ControlSessionTcp::ControlSessionTcp(ControlMessageReceiver* receiver, tcp::socket&& socket) : ControlSession(receiver), socket_(std::move(socket))
{
}
@ -42,7 +40,7 @@ void ControlSessionTcp::do_read()
{
const std::string delimiter = "\n";
auto self(shared_from_this());
asio::async_read_until(socket_, streambuf_, delimiter, [this, self, delimiter](const std::error_code& ec, std::size_t bytes_transferred) {
boost::asio::async_read_until(socket_, streambuf_, delimiter, [this, self, delimiter](const std::error_code& ec, std::size_t bytes_transferred) {
if (ec)
{
LOG(ERROR) << "Error while reading from control socket: " << ec.message() << "\n";
@ -73,8 +71,8 @@ void ControlSessionTcp::start()
void ControlSessionTcp::stop()
{
LOG(DEBUG) << "ControlSession::stop\n";
std::error_code ec;
socket_.shutdown(asio::ip::tcp::socket::shutdown_both, ec);
boost::system::error_code ec;
socket_.shutdown(boost::asio::ip::tcp::socket::shutdown_both, ec);
if (ec)
LOG(ERROR) << "Error in socket shutdown: " << ec.message() << "\n";
socket_.close(ec);
@ -87,7 +85,7 @@ void ControlSessionTcp::stop()
void ControlSessionTcp::sendAsync(const std::string& message)
{
auto self(shared_from_this());
asio::async_write(socket_, asio::buffer(message + "\r\n"), [this, self](std::error_code ec, std::size_t length) {
boost::asio::async_write(socket_, boost::asio::buffer(message + "\r\n"), [this, self](std::error_code ec, std::size_t length) {
if (ec)
{
LOG(ERROR) << "Error while writing to control socket: " << ec.message() << "\n";
@ -102,7 +100,7 @@ void ControlSessionTcp::sendAsync(const std::string& message)
bool ControlSessionTcp::send(const std::string& message)
{
error_code ec;
asio::write(socket_, asio::buffer(message + "\r\n"), ec);
boost::system::error_code ec;
boost::asio::write(socket_, boost::asio::buffer(message + "\r\n"), ec);
return !ec;
}

3
server/control_session_tcp.hpp
View file

@ -44,7 +44,8 @@ public:
protected:
void do_read();
asio::streambuf streambuf_;
tcp::socket socket_;
boost::asio::streambuf streambuf_;
};

4
server/control_session_ws.cpp
View file

@ -20,13 +20,11 @@
#include "aixlog.hpp"
#include "message/pcmChunk.h"
#include <iostream>
#include <mutex>
using namespace std;
ControlSessionWs::ControlSessionWs(ControlMessageReceiver* receiver, tcp::socket&& socket) : ControlSession(receiver, std::move(socket))
ControlSessionWs::ControlSessionWs(ControlMessageReceiver* receiver, tcp::socket&& socket) : ControlSession(receiver), ws_(std::move(socket))
{
}

13
server/control_session_ws.hpp
View file

@ -20,6 +20,14 @@
#define CONTROL_SESSION_WS_HPP
#include "control_session.hpp"
#include <boost/beast/core.hpp>
#include <boost/beast/websocket.hpp>
namespace beast = boost::beast; // from <boost/beast.hpp>
namespace http = beast::http; // from <boost/beast/http.hpp>
namespace websocket = beast::websocket; // from <boost/beast/websocket.hpp>
namespace net = boost::asio; // from <boost/asio.hpp>
/// Endpoint for a connected control client.
/**
@ -41,6 +49,11 @@ public:
/// Sends a message to the client (asynchronous)
void sendAsync(const std::string& message) override;
protected:
websocket::stream<beast::tcp_stream> ws_;
// beast::flat_buffer buffer_;
// beast::multi_buffer b;
};

4
server/snapServer.cpp
View file

@ -201,11 +201,11 @@ int main(int argc, char* argv[])
if (settings.bufferMs < 400)
settings.bufferMs = 400;
asio::io_context io_context;
boost::asio::io_context io_context;
std::unique_ptr<StreamServer> streamServer(new StreamServer(&io_context, settings));
streamServer->start();
auto func = [](asio::io_context* ioservice) -> void { ioservice->run(); };
auto func = [](boost::asio::io_context* ioservice) -> void { ioservice->run(); };
std::thread t(func, &io_context);
while (!g_terminated)

12
server/streamServer.cpp
View file

@ -29,7 +29,7 @@ using namespace std;
using json = nlohmann::json;
StreamServer::StreamServer(asio::io_context* io_context, const StreamServerSettings& streamServerSettings)
StreamServer::StreamServer(boost::asio::io_context* io_context, const StreamServerSettings& streamServerSettings)
: io_context_(io_context), acceptor_v4_(nullptr), acceptor_v6_(nullptr), settings_(streamServerSettings)
{
}
@ -784,14 +784,14 @@ void StreamServer::start()
try
{
acceptor_v6_ = make_shared<tcp::acceptor>(*io_context_, endpoint_v6);
error_code ec;
acceptor_v6_->set_option(asio::ip::v6_only(false), ec);
asio::ip::v6_only option;
boost::system::error_code ec;
acceptor_v6_->set_option(boost::asio::ip::v6_only(false), ec);
boost::asio::ip::v6_only option;
acceptor_v6_->get_option(option);
is_v6_only = option.value();
LOG(DEBUG) << "IPv6 only: " << is_v6_only << "\n";
}
catch (const asio::system_error& e)
catch (const boost::system::system_error& e)
{
LOG(ERROR) << "error creating TCP acceptor: " << e.what() << ", code: " << e.code() << "\n";
}
@ -803,7 +803,7 @@ void StreamServer::start()
{
acceptor_v4_ = make_shared<tcp::acceptor>(*io_context_, endpoint_v4);
}
catch (const asio::system_error& e)
catch (const boost::system::system_error& e)
{
LOG(ERROR) << "error creating TCP acceptor: " << e.what() << ", code: " << e.code() << "\n";
}

8
server/streamServer.h
View file

@ -19,7 +19,7 @@
#ifndef STREAM_SERVER_H
#define STREAM_SERVER_H
#include <asio.hpp>
#include <boost/asio.hpp>
#include <memory>
#include <mutex>
#include <set>
@ -38,7 +38,7 @@
#include "streamreader/streamManager.h"
using asio::ip::tcp;
using boost::asio::ip::tcp;
typedef std::shared_ptr<tcp::socket> socket_ptr;
typedef std::shared_ptr<StreamSession> session_ptr;
@ -69,7 +69,7 @@ struct StreamServerSettings
class StreamServer : public MessageReceiver, ControlMessageReceiver, PcmListener
{
public:
StreamServer(asio::io_context* io_context, const StreamServerSettings& streamServerSettings);
StreamServer(boost::asio::io_context* io_context, const StreamServerSettings& streamServerSettings);
virtual ~StreamServer();
void start();
@ -99,7 +99,7 @@ private:
void ProcessRequest(const jsonrpcpp::request_ptr request, jsonrpcpp::entity_ptr& response, jsonrpcpp::notification_ptr& notification) const;
mutable std::recursive_mutex sessionsMutex_;
std::set<session_ptr> sessions_;
asio::io_context* io_context_;
boost::asio::io_context* io_context_;
std::shared_ptr<tcp::acceptor> acceptor_v4_;
std::shared_ptr<tcp::acceptor> acceptor_v6_;

12
server/streamSession.cpp
View file

@ -75,11 +75,11 @@ void StreamSession::stop()
try
{
std::error_code ec;
boost::system::error_code ec;
if (socket_)
{
std::lock_guard<std::mutex> socketLock(socketMutex_);
socket_->shutdown(asio::ip::tcp::socket::shutdown_both, ec);
socket_->shutdown(boost::asio::ip::tcp::socket::shutdown_both, ec);
if (ec)
LOG(ERROR) << "Error in socket shutdown: " << ec.message() << "\n";
socket_->close(ec);
@ -114,7 +114,7 @@ void StreamSession::socketRead(void* _to, size_t _bytes)
size_t read = 0;
do
{
read += socket_->read_some(asio::buffer((char*)_to + read, _bytes - read));
read += socket_->read_some(boost::asio::buffer((char*)_to + read, _bytes - read));
} while (active_ && (read < _bytes));
}
@ -157,12 +157,12 @@ bool StreamSession::send(const msg::message_ptr& message) const
if (!socket_ || !active_)
return false;
}
asio::streambuf streambuf;
boost::asio::streambuf streambuf;
std::ostream stream(&streambuf);
tv t;
message->sent = t;
message->serialize(stream);
asio::write(*socket_.get(), streambuf);
boost::asio::write(*socket_.get(), streambuf);
// LOG(INFO) << "done: " << message->type << ", size: " << message->size << ", id: " << message->id << ", refers: " << message->refersTo << "\n";
return true;
}
@ -228,7 +228,7 @@ void StreamSession::writer()
{
try
{
asio::streambuf streambuf;
boost::asio::streambuf streambuf;
std::ostream stream(&streambuf);
shared_ptr<msg::BaseMessage> message;
while (active_)

4
server/streamSession.h
View file

@ -22,7 +22,7 @@
#include "common/queue.h"
#include "message/message.h"
#include "streamreader/streamManager.h"
#include <asio.hpp>
#include <boost/asio.hpp>
#include <atomic>
#include <condition_variable>
#include <memory>
@ -32,7 +32,7 @@
#include <thread>
using asio::ip::tcp;
using boost::asio::ip::tcp;
class StreamSession;