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

Reformat code

Browse source
This commit is contained in:
badaix 2022-12-29 19:10:38 +01:00
parent c6518a4709
commit e1c8250876
32 changed files with 505 additions and 351 deletions
Download patch file Download diff file Expand all files Collapse all files

43
client/browseZeroConf/browse_bonjour.cpp
View file

@ -179,11 +179,12 @@ bool BrowseBonjour::browse(const string& serviceName, mDNSResult& result, int /*
CHECKED(DNSServiceBrowse( CHECKED(DNSServiceBrowse(
service.get(), 0, 0, serviceName.c_str(), "local.", service.get(), 0, 0, serviceName.c_str(), "local.",
[](DNSServiceRef /*service*/, DNSServiceFlags /*flags*/, uint32_t /*interfaceIndex*/, DNSServiceErrorType errorCode, const char* serviceName, [](DNSServiceRef /*service*/, DNSServiceFlags /*flags*/, uint32_t /*interfaceIndex*/, DNSServiceErrorType errorCode, const char* serviceName,
const char* regtype, const char* replyDomain, void* context) { const char* regtype, const char* replyDomain, void* context)
auto replyCollection = static_cast<deque<mDNSReply>*>(context); {
auto replyCollection = static_cast<deque<mDNSReply>*>(context);
CHECKED(errorCode); CHECKED(errorCode);
replyCollection->push_back(mDNSReply{string(serviceName), string(regtype), string(replyDomain)}); replyCollection->push_back(mDNSReply{string(serviceName), string(regtype), string(replyDomain)});
}, },
&replyCollection)); &replyCollection));
@ -198,11 +199,12 @@ bool BrowseBonjour::browse(const string& serviceName, mDNSResult& result, int /*
CHECKED(DNSServiceResolve( CHECKED(DNSServiceResolve(
service.get(), 0, 0, reply.name.c_str(), reply.regtype.c_str(), reply.domain.c_str(), service.get(), 0, 0, reply.name.c_str(), reply.regtype.c_str(), reply.domain.c_str(),
[](DNSServiceRef /*service*/, DNSServiceFlags /*flags*/, uint32_t /*interfaceIndex*/, DNSServiceErrorType errorCode, const char* /*fullName*/, [](DNSServiceRef /*service*/, DNSServiceFlags /*flags*/, uint32_t /*interfaceIndex*/, DNSServiceErrorType errorCode, const char* /*fullName*/,
const char* hosttarget, uint16_t port, uint16_t /*txtLen*/, const unsigned char* /*txtRecord*/, void* context) { const char* hosttarget, uint16_t port, uint16_t /*txtLen*/, const unsigned char* /*txtRecord*/, void* context)
auto resultCollection = static_cast<deque<mDNSResolve>*>(context); {
auto resultCollection = static_cast<deque<mDNSResolve>*>(context);
CHECKED(errorCode); CHECKED(errorCode);
resultCollection->push_back(mDNSResolve{string(hosttarget), ntohs(port)}); resultCollection->push_back(mDNSResolve{string(hosttarget), ntohs(port)});
}, },
&resolveCollection)); &resolveCollection));
@ -220,20 +222,21 @@ bool BrowseBonjour::browse(const string& serviceName, mDNSResult& result, int /*
CHECKED(DNSServiceGetAddrInfo( CHECKED(DNSServiceGetAddrInfo(
service.get(), kDNSServiceFlagsLongLivedQuery, 0, kDNSServiceProtocol_IPv4, resolve.fullName.c_str(), service.get(), kDNSServiceFlagsLongLivedQuery, 0, kDNSServiceProtocol_IPv4, resolve.fullName.c_str(),
[](DNSServiceRef /*service*/, DNSServiceFlags /*flags*/, uint32_t interfaceIndex, DNSServiceErrorType /*errorCode*/, const char* hostname, [](DNSServiceRef /*service*/, DNSServiceFlags /*flags*/, uint32_t interfaceIndex, DNSServiceErrorType /*errorCode*/, const char* hostname,
const sockaddr* address, uint32_t /*ttl*/, void* context) { const sockaddr* address, uint32_t /*ttl*/, void* context)
auto result = static_cast<mDNSResult*>(context); {
auto result = static_cast<mDNSResult*>(context);
result->host = string(hostname); result->host = string(hostname);
result->ip_version = (address->sa_family == AF_INET) ? (IPVersion::IPv4) : (IPVersion::IPv6); result->ip_version = (address->sa_family == AF_INET) ? (IPVersion::IPv4) : (IPVersion::IPv6);
result->iface_idx = static_cast<int>(interfaceIndex); result->iface_idx = static_cast<int>(interfaceIndex);
char hostIP[NI_MAXHOST]; char hostIP[NI_MAXHOST];
char hostService[NI_MAXSERV]; char hostService[NI_MAXSERV];
if (getnameinfo(address, sizeof(*address), hostIP, sizeof(hostIP), hostService, sizeof(hostService), NI_NUMERICHOST | NI_NUMERICSERV) == 0) if (getnameinfo(address, sizeof(*address), hostIP, sizeof(hostIP), hostService, sizeof(hostService), NI_NUMERICHOST | NI_NUMERICSERV) == 0)
result->ip = string(hostIP); result->ip = string(hostIP);
else else
return; return;
result->valid = true; result->valid = true;
}, },
&resultCollection[i++])); &resultCollection[i++]));
} }

85
client/client_connection.cpp
View file

@ -52,7 +52,9 @@ PendingRequest::~PendingRequest()
void PendingRequest::setValue(std::unique_ptr<msg::BaseMessage> value) void PendingRequest::setValue(std::unique_ptr<msg::BaseMessage> value)
{ {
boost::asio::post(strand_, [this, self = shared_from_this(), val = std::move(value)]() mutable { boost::asio::post(strand_,
[this, self = shared_from_this(), val = std::move(value)]() mutable
{
timer_.cancel(); timer_.cancel();
if (handler_) if (handler_)
handler_({}, std::move(val)); handler_({}, std::move(val));
@ -67,7 +69,9 @@ uint16_t PendingRequest::id() const
void PendingRequest::startTimer(const chronos::usec& timeout) void PendingRequest::startTimer(const chronos::usec& timeout)
{ {
timer_.expires_after(timeout); timer_.expires_after(timeout);
timer_.async_wait([this, self = shared_from_this()](boost::system::error_code ec) { timer_.async_wait(
[this, self = shared_from_this()](boost::system::error_code ec)
{
if (!handler_) if (!handler_)
return; return;
if (!ec) if (!ec)
@ -202,7 +206,9 @@ void ClientConnection::sendNext()
message.msg->serialize(stream); message.msg->serialize(stream);
auto handler = message.handler; auto handler = message.handler;
boost::asio::async_write(socket_, streambuf, [this, handler](boost::system::error_code ec, std::size_t length) { boost::asio::async_write(socket_, streambuf,
[this, handler](boost::system::error_code ec, std::size_t length)
{
if (ec) if (ec)
LOG(ERROR, LOG_TAG) << "Failed to send message, error: " << ec.message() << "\n"; LOG(ERROR, LOG_TAG) << "Failed to send message, error: " << ec.message() << "\n";
else else
@ -220,7 +226,9 @@ void ClientConnection::sendNext()
void ClientConnection::send(const msg::message_ptr& message, const ResultHandler& handler) void ClientConnection::send(const msg::message_ptr& message, const ResultHandler& handler)
{ {
boost::asio::post(strand_, [this, message, handler]() { boost::asio::post(strand_,
[this, message, handler]()
{
messages_.emplace_back(message, handler); messages_.emplace_back(message, handler);
if (messages_.size() > 1) if (messages_.size() > 1)
{ {
@ -234,7 +242,9 @@ void ClientConnection::send(const msg::message_ptr& message, const ResultHandler
void ClientConnection::sendRequest(const msg::message_ptr& message, const chronos::usec& timeout, const MessageHandler<msg::BaseMessage>& handler) void ClientConnection::sendRequest(const msg::message_ptr& message, const chronos::usec& timeout, const MessageHandler<msg::BaseMessage>& handler)
{ {
boost::asio::post(strand_, [this, message, timeout, handler]() { boost::asio::post(strand_,
[this, message, timeout, handler]()
{
pendingRequests_.erase( pendingRequests_.erase(
std::remove_if(pendingRequests_.begin(), pendingRequests_.end(), [](std::weak_ptr<PendingRequest> request) { return request.expired(); }), std::remove_if(pendingRequests_.begin(), pendingRequests_.end(), [](std::weak_ptr<PendingRequest> request) { return request.expired(); }),
pendingRequests_.end()); pendingRequests_.end());
@ -245,7 +255,9 @@ void ClientConnection::sendRequest(const msg::message_ptr& message, const chrono
auto request = make_shared<PendingRequest>(strand_, reqId_, handler); auto request = make_shared<PendingRequest>(strand_, reqId_, handler);
pendingRequests_.push_back(request); pendingRequests_.push_back(request);
request->startTimer(timeout); request->startTimer(timeout);
send(message, [handler](const boost::system::error_code& ec) { send(message,
[handler](const boost::system::error_code& ec)
{
if (ec) if (ec)
handler(ec, nullptr); handler(ec, nullptr);
}); });
@ -255,7 +267,9 @@ void ClientConnection::sendRequest(const msg::message_ptr& message, const chrono
void ClientConnection::getNextMessage(const MessageHandler<msg::BaseMessage>& handler) void ClientConnection::getNextMessage(const MessageHandler<msg::BaseMessage>& handler)
{ {
boost::asio::async_read(socket_, boost::asio::buffer(buffer_, base_msg_size_), [this, handler](boost::system::error_code ec, std::size_t length) mutable { boost::asio::async_read(socket_, boost::asio::buffer(buffer_, base_msg_size_),
[this, handler](boost::system::error_code ec, std::size_t length) mutable
{
if (ec) if (ec)
{ {
LOG(ERROR, LOG_TAG) << "Error reading message header of length " << length << ": " << ec.message() << "\n"; LOG(ERROR, LOG_TAG) << "Error reading message header of length " << length << ": " << ec.message() << "\n";
@ -288,35 +302,36 @@ void ClientConnection::getNextMessage(const MessageHandler<msg::BaseMessage>& ha
buffer_.resize(base_message_.size); buffer_.resize(base_message_.size);
boost::asio::async_read(socket_, boost::asio::buffer(buffer_, base_message_.size), boost::asio::async_read(socket_, boost::asio::buffer(buffer_, base_message_.size),
[this, handler](boost::system::error_code ec, std::size_t length) mutable { [this, handler](boost::system::error_code ec, std::size_t length) mutable
if (ec) {
{ if (ec)
LOG(ERROR, LOG_TAG) << "Error reading message body of length " << length << ": " << ec.message() << "\n"; {
if (handler) LOG(ERROR, LOG_TAG) << "Error reading message body of length " << length << ": " << ec.message() << "\n";
handler(ec, nullptr); if (handler)
return; handler(ec, nullptr);
} return;
}
auto response = msg::factory::createMessage(base_message_, buffer_.data()); auto response = msg::factory::createMessage(base_message_, buffer_.data());
if (!response) if (!response)
LOG(WARNING, LOG_TAG) << "Failed to deserialize message of type: " << base_message_.type << "\n"; LOG(WARNING, LOG_TAG) << "Failed to deserialize message of type: " << base_message_.type << "\n";
for (auto iter = pendingRequests_.begin(); iter != pendingRequests_.end(); ++iter) for (auto iter = pendingRequests_.begin(); iter != pendingRequests_.end(); ++iter)
{ {
auto request = *iter; auto request = *iter;
if (auto req = request.lock()) if (auto req = request.lock())
{ {
if (req->id() == base_message_.refersTo) if (req->id() == base_message_.refersTo)
{ {
req->setValue(std::move(response)); req->setValue(std::move(response));
pendingRequests_.erase(iter); pendingRequests_.erase(iter);
getNextMessage(handler); getNextMessage(handler);
return; return;
} }
} }
} }
if (handler) if (handler)
handler(ec, std::move(response)); handler(ec, std::move(response));
}); });
}); });
} }

4
client/client_connection.hpp
View file

@ -117,7 +117,9 @@ public:
template <typename Message> template <typename Message>
void sendRequest(const msg::message_ptr& message, const chronos::usec& timeout, const MessageHandler<Message>& handler) void sendRequest(const msg::message_ptr& message, const chronos::usec& timeout, const MessageHandler<Message>& handler)
{ {
sendRequest(message, timeout, [handler](const boost::system::error_code& ec, std::unique_ptr<msg::BaseMessage> response) { sendRequest(message, timeout,
[handler](const boost::system::error_code& ec, std::unique_ptr<msg::BaseMessage> response)
{
if (ec) if (ec)
handler(ec, nullptr); handler(ec, nullptr);
else if (auto casted_response = msg::message_cast<Message>(std::move(response))) else if (auto casted_response = msg::message_cast<Message>(std::move(response)))

109
client/controller.cpp
View file

@ -121,7 +121,9 @@ std::vector<std::string> Controller::getSupportedPlayerNames()
void Controller::getNextMessage() void Controller::getNextMessage()
{ {
clientConnection_->getNextMessage([this](const boost::system::error_code& ec, std::unique_ptr<msg::BaseMessage> response) { clientConnection_->getNextMessage(
[this](const boost::system::error_code& ec, std::unique_ptr<msg::BaseMessage> response)
{
if (ec) if (ec)
{ {
reconnect(); reconnect();
@ -222,7 +224,9 @@ void Controller::getNextMessage()
if (!player_) if (!player_)
throw SnapException("No audio player support" + (settings_.player.player_name.empty() ? "" : " for: " + settings_.player.player_name)); throw SnapException("No audio player support" + (settings_.player.player_name.empty() ? "" : " for: " + settings_.player.player_name));
player_->setVolumeCallback([this](double volume, bool muted) { player_->setVolumeCallback(
[this](double volume, bool muted)
{
static double last_volume(-1); static double last_volume(-1);
static bool last_muted(true); static bool last_muted(true);
if ((volume != last_volume) || (last_muted != muted)) if ((volume != last_volume) || (last_muted != muted))
@ -232,7 +236,9 @@ void Controller::getNextMessage()
auto info = std::make_shared<msg::ClientInfo>(); auto info = std::make_shared<msg::ClientInfo>();
info->setVolume(static_cast<uint16_t>(volume * 100.)); info->setVolume(static_cast<uint16_t>(volume * 100.));
info->setMuted(muted); info->setMuted(muted);
clientConnection_->send(info, [this](const boost::system::error_code& ec) { clientConnection_->send(info,
[this](const boost::system::error_code& ec)
{
if (ec) if (ec)
{ {
LOG(ERROR, LOG_TAG) << "Failed to send client info, error: " << ec.message() << "\n"; LOG(ERROR, LOG_TAG) << "Failed to send client info, error: " << ec.message() << "\n";
@ -270,35 +276,38 @@ void Controller::getNextMessage()
void Controller::sendTimeSyncMessage(int quick_syncs) void Controller::sendTimeSyncMessage(int quick_syncs)
{ {
auto timeReq = std::make_shared<msg::Time>(); auto timeReq = std::make_shared<msg::Time>();
clientConnection_->sendRequest<msg::Time>( clientConnection_->sendRequest<msg::Time>(timeReq, 2s,
timeReq, 2s, [this, quick_syncs](const boost::system::error_code& ec, const std::unique_ptr<msg::Time>& response) mutable { [this, quick_syncs](const boost::system::error_code& ec, const std::unique_ptr<msg::Time>& response) mutable
if (ec) {
{ if (ec)
LOG(ERROR, LOG_TAG) << "Time sync request failed: " << ec.message() << "\n"; {
reconnect(); LOG(ERROR, LOG_TAG) << "Time sync request failed: " << ec.message() << "\n";
return; reconnect();
} return;
else }
{ else
TimeProvider::getInstance().setDiff(response->latency, response->received - response->sent); {
} TimeProvider::getInstance().setDiff(response->latency, response->received - response->sent);
}
std::chrono::microseconds next = TIME_SYNC_INTERVAL; std::chrono::microseconds next = TIME_SYNC_INTERVAL;
if (quick_syncs > 0) if (quick_syncs > 0)
{
if (--quick_syncs == 0)
LOG(INFO, LOG_TAG) << "diff to server [ms]: "
<< static_cast<float>(TimeProvider::getInstance().getDiffToServer<chronos::usec>().count()) / 1000.f << "\n";
next = 100us;
}
timer_.expires_after(next);
timer_.async_wait(
[this, quick_syncs](const boost::system::error_code& ec)
{ {
if (--quick_syncs == 0) if (!ec)
LOG(INFO, LOG_TAG) << "diff to server [ms]: " {
<< static_cast<float>(TimeProvider::getInstance().getDiffToServer<chronos::usec>().count()) / 1000.f << "\n"; sendTimeSyncMessage(quick_syncs);
next = 100us;
} }
timer_.expires_after(next);
timer_.async_wait([this, quick_syncs](const boost::system::error_code& ec) {
if (!ec)
{
sendTimeSyncMessage(quick_syncs);
}
});
}); });
});
} }
void Controller::browseMdns(const MdnsHandler& handler) void Controller::browseMdns(const MdnsHandler& handler)
@ -324,7 +333,9 @@ void Controller::browseMdns(const MdnsHandler& handler)
} }
timer_.expires_after(500ms); timer_.expires_after(500ms);
timer_.async_wait([this, handler](const boost::system::error_code& ec) { timer_.async_wait(
[this, handler](const boost::system::error_code& ec)
{
if (!ec) if (!ec)
{ {
browseMdns(handler); browseMdns(handler);
@ -343,7 +354,9 @@ void Controller::start()
{ {
if (settings_.server.host.empty()) if (settings_.server.host.empty())
{ {
browseMdns([this](const boost::system::error_code& ec, const std::string& host, uint16_t port) { browseMdns(
[this](const boost::system::error_code& ec, const std::string& host, uint16_t port)
{
if (ec) if (ec)
{ {
LOG(ERROR, LOG_TAG) << "Failed to browse MDNS, error: " << ec.message() << "\n"; LOG(ERROR, LOG_TAG) << "Failed to browse MDNS, error: " << ec.message() << "\n";
@ -380,7 +393,9 @@ void Controller::reconnect()
stream_.reset(); stream_.reset();
decoder_.reset(); decoder_.reset();
timer_.expires_after(1s); timer_.expires_after(1s);
timer_.async_wait([this](const boost::system::error_code& ec) { timer_.async_wait(
[this](const boost::system::error_code& ec)
{
if (!ec) if (!ec)
{ {
worker(); worker();
@ -390,7 +405,9 @@ void Controller::reconnect()
void Controller::worker() void Controller::worker()
{ {
clientConnection_->connect([this](const boost::system::error_code& ec) { clientConnection_->connect(
[this](const boost::system::error_code& ec)
{
if (!ec) if (!ec)
{ {
// LOG(INFO, LOG_TAG) << "Connected!\n"; // LOG(INFO, LOG_TAG) << "Connected!\n";
@ -401,19 +418,21 @@ void Controller::worker()
// Say hello to the server // Say hello to the server
auto hello = std::make_shared<msg::Hello>(macAddress, settings_.host_id, settings_.instance); auto hello = std::make_shared<msg::Hello>(macAddress, settings_.host_id, settings_.instance);
clientConnection_->sendRequest<msg::ServerSettings>( clientConnection_->sendRequest<msg::ServerSettings>(
hello, 2s, [this](const boost::system::error_code& ec, std::unique_ptr<msg::ServerSettings> response) mutable { hello, 2s,
if (ec) [this](const boost::system::error_code& ec, std::unique_ptr<msg::ServerSettings> response) mutable
{ {
LOG(ERROR, LOG_TAG) << "Failed to send hello request, error: " << ec.message() << "\n"; if (ec)
reconnect(); {
return; LOG(ERROR, LOG_TAG) << "Failed to send hello request, error: " << ec.message() << "\n";
} reconnect();
else return;
{ }
serverSettings_ = std::move(response); else
LOG(INFO, LOG_TAG) << "ServerSettings - buffer: " << serverSettings_->getBufferMs() << ", latency: " << serverSettings_->getLatency() {
<< ", volume: " << serverSettings_->getVolume() << ", muted: " << serverSettings_->isMuted() << "\n"; serverSettings_ = std::move(response);
} LOG(INFO, LOG_TAG) << "ServerSettings - buffer: " << serverSettings_->getBufferMs() << ", latency: " << serverSettings_->getLatency()
<< ", volume: " << serverSettings_->getVolume() << ", muted: " << serverSettings_->isMuted() << "\n";
}
}); });
// Do initial time sync with the server // Do initial time sync with the server

12
client/player/alsa_player.cpp
View file

@ -188,7 +188,9 @@ bool AlsaPlayer::getHardwareVolume(double& volume, bool& muted)
void AlsaPlayer::waitForEvent() void AlsaPlayer::waitForEvent()
{ {
sd_.async_wait(boost::asio::posix::stream_descriptor::wait_read, [this](const boost::system::error_code& ec) { sd_.async_wait(boost::asio::posix::stream_descriptor::wait_read,
[this](const boost::system::error_code& ec)
{
if (ec) if (ec)
{ {
// TODO: fd is "Bad" after unplugging/plugging USB DAC, i.e. after init/uninit/init cycle // TODO: fd is "Bad" after unplugging/plugging USB DAC, i.e. after init/uninit/init cycle
@ -221,7 +223,9 @@ void AlsaPlayer::waitForEvent()
// As workaround we defer getting the volume by 20ms. // As workaround we defer getting the volume by 20ms.
timer_.cancel(); timer_.cancel();
timer_.expires_after(20ms); timer_.expires_after(20ms);
timer_.async_wait([this](const boost::system::error_code& ec) { timer_.async_wait(
[this](const boost::system::error_code& ec)
{
if (!ec) if (!ec)
{ {
if (getHardwareVolume(volume_, muted_)) if (getHardwareVolume(volume_, muted_))
@ -249,7 +253,9 @@ void AlsaPlayer::initMixer()
throw SnapException("Can't open control for " + mixer_device_ + ", error: " + snd_strerror(err)); throw SnapException("Can't open control for " + mixer_device_ + ", error: " + snd_strerror(err));
if ((err = snd_ctl_subscribe_events(ctl_, 1)) < 0) if ((err = snd_ctl_subscribe_events(ctl_, 1)) < 0)
throw SnapException("Can't subscribe for events for " + mixer_device_ + ", error: " + snd_strerror(err)); throw SnapException("Can't subscribe for events for " + mixer_device_ + ", error: " + snd_strerror(err));
fd_ = std::unique_ptr<pollfd, std::function<void(pollfd*)>>(new pollfd(), [](pollfd* p) { fd_ = std::unique_ptr<pollfd, std::function<void(pollfd*)>>(new pollfd(),
[](pollfd* p)
{
close(p->fd); close(p->fd);
delete p; delete p;
}); });

4
client/player/file_player.cpp
View file

@ -131,7 +131,9 @@ void FilePlayer::loop()
next_request_ = now + 1ms; next_request_ = now + 1ms;
timer_.expires_at(next_request_); timer_.expires_at(next_request_);
timer_.async_wait([this](boost::system::error_code ec) { timer_.async_wait(
[this](boost::system::error_code ec)
{
if (ec) if (ec)
return; return;
requestAudio(); requestAudio();

3
client/player/player.cpp
View file

@ -71,7 +71,8 @@ Player::Player(boost::asio::io_context& io_context, const ClientSettings::Player
break; break;
} }
auto not_empty = [](const std::string& value) -> std::string { auto not_empty = [](const std::string& value) -> std::string
{
if (!value.empty()) if (!value.empty())
return value; return value;
else else

130
client/player/pulse_player.cpp
View file

@ -54,7 +54,9 @@ vector<PcmDevice> PulsePlayer::pcm_list(const std::string& parameter)
{ {
auto pa_ml = std::shared_ptr<pa_mainloop>(pa_mainloop_new(), [](pa_mainloop* pa_ml) { pa_mainloop_free(pa_ml); }); auto pa_ml = std::shared_ptr<pa_mainloop>(pa_mainloop_new(), [](pa_mainloop* pa_ml) { pa_mainloop_free(pa_ml); });
pa_mainloop_api* pa_mlapi = pa_mainloop_get_api(pa_ml.get()); pa_mainloop_api* pa_mlapi = pa_mainloop_get_api(pa_ml.get());
auto pa_ctx = std::shared_ptr<pa_context>(pa_context_new(pa_mlapi, "Snapcast"), [](pa_context* pa_ctx) { auto pa_ctx = std::shared_ptr<pa_context>(pa_context_new(pa_mlapi, "Snapcast"),
[](pa_context* pa_ctx)
{
pa_context_disconnect(pa_ctx); pa_context_disconnect(pa_ctx);
pa_context_unref(pa_ctx); pa_context_unref(pa_ctx);
}); });
@ -70,21 +72,22 @@ vector<PcmDevice> PulsePlayer::pcm_list(const std::string& parameter)
static int pa_ready = 0; static int pa_ready = 0;
pa_context_set_state_callback( pa_context_set_state_callback(
pa_ctx.get(), pa_ctx.get(),
[](pa_context* c, void* userdata) { [](pa_context* c, void* userdata)
std::ignore = userdata; {
pa_context_state_t state = pa_context_get_state(c); std::ignore = userdata;
switch (state) pa_context_state_t state = pa_context_get_state(c);
{ switch (state)
case PA_CONTEXT_FAILED: {
case PA_CONTEXT_TERMINATED: case PA_CONTEXT_FAILED:
pa_ready = 2; case PA_CONTEXT_TERMINATED:
break; pa_ready = 2;
case PA_CONTEXT_READY: break;
pa_ready = 1; case PA_CONTEXT_READY:
break; pa_ready = 1;
default: break;
break; default:
} break;
}
}, },
nullptr); nullptr);
@ -107,13 +110,14 @@ vector<PcmDevice> PulsePlayer::pcm_list(const std::string& parameter)
static std::vector<PcmDevice> devices; static std::vector<PcmDevice> devices;
auto* op = pa_context_get_sink_info_list( auto* op = pa_context_get_sink_info_list(
pa_ctx.get(), pa_ctx.get(),
[](pa_context* ctx, const pa_sink_info* i, int eol, void* userdata) mutable { [](pa_context* ctx, const pa_sink_info* i, int eol, void* userdata) mutable
std::ignore = ctx; {
std::ignore = userdata; std::ignore = ctx;
// auto self = static_cast<PulsePlayer*>(userdata); std::ignore = userdata;
// If eol is set to a positive number, you're at the end of the list // auto self = static_cast<PulsePlayer*>(userdata);
if (eol <= 0) // If eol is set to a positive number, you're at the end of the list
devices.emplace_back(i->index, i->name, i->description); if (eol <= 0)
devices.emplace_back(i->index, i->name, i->description);
}, },
nullptr); nullptr);
@ -246,26 +250,27 @@ void PulsePlayer::triggerVolumeUpdate()
{ {
pa_context_get_sink_input_info( pa_context_get_sink_input_info(
pa_ctx_, pa_stream_get_index(playstream_), pa_ctx_, pa_stream_get_index(playstream_),
[](pa_context* ctx, const pa_sink_input_info* info, int eol, void* userdata) { [](pa_context* ctx, const pa_sink_input_info* info, int eol, void* userdata)
std::ignore = ctx; {
LOG(DEBUG, LOG_TAG) << "pa_context_get_sink_info_by_index info: " << (info != nullptr) << ", eol: " << eol << "\n"; std::ignore = ctx;
if (info != nullptr) LOG(DEBUG, LOG_TAG) << "pa_context_get_sink_info_by_index info: " << (info != nullptr) << ", eol: " << eol << "\n";
{ if (info != nullptr)
auto* self = static_cast<PulsePlayer*>(userdata); {
self->volume_ = static_cast<double>(pa_cvolume_avg(&(info->volume))) / static_cast<double>(PA_VOLUME_NORM); auto* self = static_cast<PulsePlayer*>(userdata);
self->muted_ = (info->mute != 0); self->volume_ = static_cast<double>(pa_cvolume_avg(&(info->volume))) / static_cast<double>(PA_VOLUME_NORM);
LOG(DEBUG, LOG_TAG) << "Volume changed: " << self->volume_ << ", muted: " << self->muted_ << "\n"; self->muted_ = (info->mute != 0);
LOG(DEBUG, LOG_TAG) << "Volume changed: " << self->volume_ << ", muted: " << self->muted_ << "\n";
auto now = std::chrono::steady_clock::now(); auto now = std::chrono::steady_clock::now();
if (now - self->last_change_ < 1s) if (now - self->last_change_ < 1s)
{ {
LOG(DEBUG, LOG_TAG) << "Last volume change by server: " LOG(DEBUG, LOG_TAG) << "Last volume change by server: "
<< std::chrono::duration_cast<std::chrono::milliseconds>(now - self->last_change_).count() << std::chrono::duration_cast<std::chrono::milliseconds>(now - self->last_change_).count()
<< " ms => ignoring volume change\n"; << " ms => ignoring volume change\n";
return; return;
}
self->notifyVolumeChange(self->volume_, self->muted_);
} }
self->notifyVolumeChange(self->volume_, self->muted_);
}
}, },
this); this);
} }
@ -432,9 +437,10 @@ void PulsePlayer::connect()
// If there's an error, the callback will set pa_ready to 2 // If there's an error, the callback will set pa_ready to 2
pa_context_set_state_callback( pa_context_set_state_callback(
pa_ctx_, pa_ctx_,
[](pa_context* c, void* userdata) { [](pa_context* c, void* userdata)
auto* self = static_cast<PulsePlayer*>(userdata); {
self->stateCallback(c); auto* self = static_cast<PulsePlayer*>(userdata);
self->stateCallback(c);
}, },
this); this);
@ -462,39 +468,43 @@ void PulsePlayer::connect()
{ {
pa_context_set_subscribe_callback( pa_context_set_subscribe_callback(
pa_ctx_, pa_ctx_,
[](pa_context* ctx, pa_subscription_event_type_t event_type, uint32_t idx, void* userdata) { [](pa_context* ctx, pa_subscription_event_type_t event_type, uint32_t idx, void* userdata)
auto* self = static_cast<PulsePlayer*>(userdata); {
self->subscribeCallback(ctx, event_type, idx); auto* self = static_cast<PulsePlayer*>(userdata);
self->subscribeCallback(ctx, event_type, idx);
}, },
this); this);
const auto mask = static_cast<pa_subscription_mask_t>(PA_SUBSCRIPTION_MASK_SINK_INPUT); const auto mask = static_cast<pa_subscription_mask_t>(PA_SUBSCRIPTION_MASK_SINK_INPUT);
pa_context_subscribe( pa_context_subscribe(
pa_ctx_, mask, pa_ctx_, mask,
[](pa_context* ctx, int success, void* userdata) { [](pa_context* ctx, int success, void* userdata)
std::ignore = ctx; {
if (success != 0) std::ignore = ctx;
{ if (success != 0)
auto* self = static_cast<PulsePlayer*>(userdata); {
self->triggerVolumeUpdate(); auto* self = static_cast<PulsePlayer*>(userdata);
} self->triggerVolumeUpdate();
}
}, },
this); this);
} }
pa_stream_set_write_callback( pa_stream_set_write_callback(
playstream_, playstream_,
[](pa_stream* stream, size_t length, void* userdata) { [](pa_stream* stream, size_t length, void* userdata)
auto* self = static_cast<PulsePlayer*>(userdata); {
self->writeCallback(stream, length); auto* self = static_cast<PulsePlayer*>(userdata);
self->writeCallback(stream, length);
}, },
this); this);
pa_stream_set_underflow_callback( pa_stream_set_underflow_callback(
playstream_, playstream_,
[](pa_stream* stream, void* userdata) { [](pa_stream* stream, void* userdata)
auto* self = static_cast<PulsePlayer*>(userdata); {
self->underflowCallback(stream); auto* self = static_cast<PulsePlayer*>(userdata);
self->underflowCallback(stream);
}, },
this); this);

3
client/player/wasapi_player.cpp
View file

@ -134,7 +134,8 @@ inline PcmDevice convertToDevice(int idx, IMMDevicePtr& device)
desc.idx = idx; desc.idx = idx;
// Convert a wide Unicode string to an UTF8 string // Convert a wide Unicode string to an UTF8 string
auto utf8_encode = [](const std::wstring& wstr) { auto utf8_encode = [](const std::wstring& wstr)
{
if (wstr.empty()) if (wstr.empty())
return std::string(); return std::string();
int size_needed = WideCharToMultiByte(CP_UTF8, 0, &wstr[0], (int)wstr.size(), NULL, 0, NULL, NULL); int size_needed = WideCharToMultiByte(CP_UTF8, 0, &wstr[0], (int)wstr.size(), NULL, 0, NULL, NULL);

4
client/snapclient.cpp
View file

@ -418,7 +418,9 @@ int main(int argc, char** argv)
boost::asio::io_context io_context; boost::asio::io_context io_context;
// Construct a signal set registered for process termination. // Construct a signal set registered for process termination.
boost::asio::signal_set signals(io_context, SIGHUP, SIGINT, SIGTERM); boost::asio::signal_set signals(io_context, SIGHUP, SIGINT, SIGTERM);
signals.async_wait([&](const boost::system::error_code& ec, int signal) { signals.async_wait(
[&](const boost::system::error_code& ec, int signal)
{
if (!ec) if (!ec)
LOG(INFO, LOG_TAG) << "Received signal " << signal << ": " << strsignal(signal) << "\n"; LOG(INFO, LOG_TAG) << "Received signal " << signal << ": " << strsignal(signal) << "\n";
else else

9
common/popl.hpp
View file

@ -945,18 +945,21 @@ inline void OptionParser::parse(const std::string& ini_filename)
std::ifstream file(ini_filename.c_str()); std::ifstream file(ini_filename.c_str());
std::string line; std::string line;
auto trim = [](std::string& s) { auto trim = [](std::string& s)
{
s.erase(s.begin(), std::find_if(s.begin(), s.end(), [](int ch) { return !std::isspace(ch); })); s.erase(s.begin(), std::find_if(s.begin(), s.end(), [](int ch) { return !std::isspace(ch); }));
s.erase(std::find_if(s.rbegin(), s.rend(), [](int ch) { return !std::isspace(ch); }).base(), s.end()); s.erase(std::find_if(s.rbegin(), s.rend(), [](int ch) { return !std::isspace(ch); }).base(), s.end());
return s; return s;
}; };
auto trim_copy = [trim](const std::string& s) { auto trim_copy = [trim](const std::string& s)
{
std::string copy(s); std::string copy(s);
return trim(copy); return trim(copy);
}; };
auto split = [trim_copy](const std::string& s) -> std::pair<std::string, std::string> { auto split = [trim_copy](const std::string& s) -> std::pair<std::string, std::string>
{
size_t pos = s.find('='); size_t pos = s.find('=');
if (pos == std::string::npos) if (pos == std::string::npos)
return {"", ""}; return {"", ""};

4
common/utils.hpp
View file

@ -72,7 +72,9 @@ namespace strutils = utils::string;
#ifndef WINDOWS #ifndef WINDOWS
static std::string execGetOutput(const std::string& cmd) static std::string execGetOutput(const std::string& cmd)
{ {
std::shared_ptr<::FILE> pipe(popen((cmd + " 2> /dev/null").c_str(), "r"), [](::FILE* stream) { std::shared_ptr<::FILE> pipe(popen((cmd + " 2> /dev/null").c_str(), "r"),
[](::FILE* stream)
{
if (stream != nullptr) if (stream != nullptr)
pclose(stream); pclose(stream);
}); });

6
server/control_server.cpp
View file

@ -106,14 +106,16 @@ void ControlServer::onNewSession(std::shared_ptr<StreamSession> session)
void ControlServer::startAccept() void ControlServer::startAccept()
{ {
auto accept_handler_tcp = [this](error_code ec, tcp::socket socket) { auto accept_handler_tcp = [this](error_code ec, tcp::socket socket)
{
if (!ec) if (!ec)
handleAccept<ControlSessionTcp>(std::move(socket)); handleAccept<ControlSessionTcp>(std::move(socket));
else else
LOG(ERROR, LOG_TAG) << "Error while accepting socket connection: " << ec.message() << "\n"; LOG(ERROR, LOG_TAG) << "Error while accepting socket connection: " << ec.message() << "\n";
}; };
auto accept_handler_http = [this](error_code ec, tcp::socket socket) { auto accept_handler_http = [this](error_code ec, tcp::socket socket)
{
if (!ec) if (!ec)
handleAccept<ControlSessionHttp>(std::move(socket), http_settings_); handleAccept<ControlSessionHttp>(std::move(socket), http_settings_);
else else

31
server/control_session_http.cpp
View file

@ -71,7 +71,8 @@ namespace
boost::beast::string_view mime_type(boost::beast::string_view path) boost::beast::string_view mime_type(boost::beast::string_view path)
{ {
using boost::beast::iequals; using boost::beast::iequals;
auto const ext = [&path] { auto const ext = [&path]
{
auto const pos = path.rfind("."); auto const pos = path.rfind(".");
if (pos == boost::beast::string_view::npos) if (pos == boost::beast::string_view::npos)
return boost::beast::string_view{}; return boost::beast::string_view{};
@ -165,7 +166,8 @@ template <class Body, class Allocator, class Send>
void ControlSessionHttp::handle_request(http::request<Body, http::basic_fields<Allocator>>&& req, Send&& send) void ControlSessionHttp::handle_request(http::request<Body, http::basic_fields<Allocator>>&& req, Send&& send)
{ {
// Returns a bad request response // Returns a bad request response
auto const bad_request = [&req](boost::beast::string_view why) { auto const bad_request = [&req](boost::beast::string_view why)
{
http::response<http::string_body> res{http::status::bad_request, req.version()}; http::response<http::string_body> res{http::status::bad_request, req.version()};
// TODO: Server: Snapcast/VERSION // TODO: Server: Snapcast/VERSION
res.set(http::field::server, HTTP_SERVER_NAME); res.set(http::field::server, HTTP_SERVER_NAME);
@ -177,7 +179,8 @@ void ControlSessionHttp::handle_request(http::request<Body, http::basic_fields<A
}; };
// Returns a not found response // Returns a not found response
auto const not_found = [&req](boost::beast::string_view target) { auto const not_found = [&req](boost::beast::string_view target)
{
http::response<http::string_body> res{http::status::not_found, req.version()}; http::response<http::string_body> res{http::status::not_found, req.version()};
res.set(http::field::server, HTTP_SERVER_NAME); res.set(http::field::server, HTTP_SERVER_NAME);
res.set(http::field::content_type, "text/html"); res.set(http::field::content_type, "text/html");
@ -188,7 +191,8 @@ void ControlSessionHttp::handle_request(http::request<Body, http::basic_fields<A
}; };
// Returns a configuration help // Returns a configuration help
auto const unconfigured = [&req]() { auto const unconfigured = [&req]()
{
http::response<http::string_body> res{http::status::ok, req.version()}; http::response<http::string_body> res{http::status::ok, req.version()};
res.set(http::field::server, HTTP_SERVER_NAME); res.set(http::field::server, HTTP_SERVER_NAME);
res.set(http::field::content_type, "text/html"); res.set(http::field::content_type, "text/html");
@ -199,7 +203,8 @@ void ControlSessionHttp::handle_request(http::request<Body, http::basic_fields<A
}; };
// Returns a server error response // Returns a server error response
auto const server_error = [&req](boost::beast::string_view what) { auto const server_error = [&req](boost::beast::string_view what)
{
http::response<http::string_body> res{http::status::internal_server_error, req.version()}; http::response<http::string_body> res{http::status::internal_server_error, req.version()};
res.set(http::field::server, HTTP_SERVER_NAME); res.set(http::field::server, HTTP_SERVER_NAME);
res.set(http::field::content_type, "text/html"); res.set(http::field::content_type, "text/html");
@ -220,7 +225,9 @@ void ControlSessionHttp::handle_request(http::request<Body, http::basic_fields<A
return send(bad_request("Illegal request-target")); return send(bad_request("Illegal request-target"));
std::string request = req.body(); std::string request = req.body();
return message_receiver_->onMessageReceived(shared_from_this(), request, [req = std::move(req), send = std::move(send)](const std::string& response) { return message_receiver_->onMessageReceived(shared_from_this(), request,
[req = std::move(req), send = std::move(send)](const std::string& response)
{
http::response<http::string_body> res{http::status::ok, req.version()}; http::response<http::string_body> res{http::status::ok, req.version()};
res.set(http::field::server, HTTP_SERVER_NAME); res.set(http::field::server, HTTP_SERVER_NAME);
res.set(http::field::content_type, "application/json"); res.set(http::field::content_type, "application/json");
@ -340,7 +347,9 @@ void ControlSessionHttp::on_read(beast::error_code ec, std::size_t bytes_transfe
// Create a WebSocket session by transferring the socket // Create a WebSocket session by transferring the socket
// std::make_shared<websocket_session>(std::move(socket_), state_)->run(std::move(req_)); // std::make_shared<websocket_session>(std::move(socket_), state_)->run(std::move(req_));
auto ws = std::make_shared<websocket::stream<beast::tcp_stream>>(std::move(socket_)); auto ws = std::make_shared<websocket::stream<beast::tcp_stream>>(std::move(socket_));
ws->async_accept(req_, [this, ws, self = shared_from_this()](beast::error_code ec) { ws->async_accept(req_,
[this, ws, self = shared_from_this()](beast::error_code ec)
{
if (ec) if (ec)
{ {
LOG(ERROR, LOG_TAG) << "Error during WebSocket handshake (control): " << ec.message() << "\n"; LOG(ERROR, LOG_TAG) << "Error during WebSocket handshake (control): " << ec.message() << "\n";
@ -357,7 +366,9 @@ void ControlSessionHttp::on_read(beast::error_code ec, std::size_t bytes_transfe
// Create a WebSocket session by transferring the socket // Create a WebSocket session by transferring the socket
// std::make_shared<websocket_session>(std::move(socket_), state_)->run(std::move(req_)); // std::make_shared<websocket_session>(std::move(socket_), state_)->run(std::move(req_));
auto ws = std::make_shared<websocket::stream<beast::tcp_stream>>(std::move(socket_)); auto ws = std::make_shared<websocket::stream<beast::tcp_stream>>(std::move(socket_));
ws->async_accept(req_, [this, ws, self = shared_from_this()](beast::error_code ec) { ws->async_accept(req_,
[this, ws, self = shared_from_this()](beast::error_code ec)
{
if (ec) if (ec)
{ {
LOG(ERROR, LOG_TAG) << "Error during WebSocket handshake (stream): " << ec.message() << "\n"; LOG(ERROR, LOG_TAG) << "Error during WebSocket handshake (stream): " << ec.message() << "\n";
@ -373,7 +384,9 @@ void ControlSessionHttp::on_read(beast::error_code ec, std::size_t bytes_transfe
} }
// Send the response // Send the response
handle_request(std::move(req_), [this](auto&& response) { handle_request(std::move(req_),
[this](auto&& response)
{
// The lifetime of the message has to extend // The lifetime of the message has to extend
// for the duration of the async operation so // for the duration of the async operation so
// we use a shared_ptr to manage it. // we use a shared_ptr to manage it.

57
server/control_session_tcp.cpp
View file

@ -51,32 +51,35 @@ ControlSessionTcp::~ControlSessionTcp()
void ControlSessionTcp::do_read() void ControlSessionTcp::do_read()
{ {
const std::string delimiter = "\n"; const std::string delimiter = "\n";
boost::asio::async_read_until( boost::asio::async_read_until(socket_, streambuf_, delimiter,
socket_, streambuf_, delimiter, [this, self = shared_from_this(), delimiter](const std::error_code& ec, std::size_t bytes_transferred) { [this, self = shared_from_this(), delimiter](const std::error_code& ec, std::size_t bytes_transferred)
if (ec) {
{ if (ec)
LOG(ERROR, LOG_TAG) << "Error while reading from control socket: " << ec.message() << "\n"; {
return; LOG(ERROR, LOG_TAG) << "Error while reading from control socket: " << ec.message() << "\n";
} return;
}
// Extract up to the first delimiter. // Extract up to the first delimiter.
std::string line{buffers_begin(streambuf_.data()), buffers_begin(streambuf_.data()) + bytes_transferred - delimiter.length()}; std::string line{buffers_begin(streambuf_.data()), buffers_begin(streambuf_.data()) + bytes_transferred - delimiter.length()};
if (!line.empty()) if (!line.empty())
{
if (line.back() == '\r')
line.resize(line.size() - 1);
// LOG(DEBUG, LOG_TAG) << "received: " << line << "\n";
if ((message_receiver_ != nullptr) && !line.empty())
{ {
if (line.back() == '\r') message_receiver_->onMessageReceived(shared_from_this(), line,
line.resize(line.size() - 1); [this](const std::string& response)
// LOG(DEBUG, LOG_TAG) << "received: " << line << "\n"; {
if ((message_receiver_ != nullptr) && !line.empty()) if (!response.empty())
{ sendAsync(response);
message_receiver_->onMessageReceived(shared_from_this(), line, [this](const std::string& response) { });
if (!response.empty())
sendAsync(response);
});
}
} }
streambuf_.consume(bytes_transferred); }
do_read(); streambuf_.consume(bytes_transferred);
}); do_read();
});
} }
@ -102,7 +105,9 @@ void ControlSessionTcp::stop()
void ControlSessionTcp::sendAsync(const std::string& message) void ControlSessionTcp::sendAsync(const std::string& message)
{ {
boost::asio::post(strand_, [this, self = shared_from_this(), message]() { boost::asio::post(strand_,
[this, self = shared_from_this(), message]()
{
messages_.emplace_back(message + "\r\n"); messages_.emplace_back(message + "\r\n");
if (messages_.size() > 1) if (messages_.size() > 1)
{ {
@ -115,7 +120,9 @@ void ControlSessionTcp::sendAsync(const std::string& message)
void ControlSessionTcp::send_next() void ControlSessionTcp::send_next()
{ {
boost::asio::async_write(socket_, boost::asio::buffer(messages_.front()), [this, self = shared_from_this()](std::error_code ec, std::size_t length) { boost::asio::async_write(socket_, boost::asio::buffer(messages_.front()),
[this, self = shared_from_this()](std::error_code ec, std::size_t length)
{
messages_.pop_front(); messages_.pop_front();
if (ec) if (ec)
{ {

12
server/control_session_ws.cpp
View file

@ -68,7 +68,9 @@ void ControlSessionWebsocket::stop()
void ControlSessionWebsocket::sendAsync(const std::string& message) void ControlSessionWebsocket::sendAsync(const std::string& message)
{ {
boost::asio::post(strand_, [this, self = shared_from_this(), msg = message]() { boost::asio::post(strand_,
[this, self = shared_from_this(), msg = message]()
{
messages_.push_back(std::move(msg)); messages_.push_back(std::move(msg));
if (messages_.size() > 1) if (messages_.size() > 1)
{ {
@ -83,7 +85,9 @@ void ControlSessionWebsocket::sendAsync(const std::string& message)
void ControlSessionWebsocket::send_next() void ControlSessionWebsocket::send_next()
{ {
const std::string& message = messages_.front(); const std::string& message = messages_.front();
ws_.async_write(boost::asio::buffer(message), [this, self = shared_from_this()](std::error_code ec, std::size_t length) { ws_.async_write(boost::asio::buffer(message),
[this, self = shared_from_this()](std::error_code ec, std::size_t length)
{
messages_.pop_front(); messages_.pop_front();
if (ec) if (ec)
{ {
@ -126,7 +130,9 @@ void ControlSessionWebsocket::on_read_ws(beast::error_code ec, std::size_t bytes
// LOG(DEBUG, LOG_TAG) << "received: " << line << "\n"; // LOG(DEBUG, LOG_TAG) << "received: " << line << "\n";
if ((message_receiver_ != nullptr) && !line.empty()) if ((message_receiver_ != nullptr) && !line.empty())
{ {
message_receiver_->onMessageReceived(shared_from_this(), line, [this](const std::string& response) { message_receiver_->onMessageReceived(shared_from_this(), line,
[this](const std::string& response)
{
if (!response.empty()) if (!response.empty())
{ {
sendAsync(response); sendAsync(response);

3
server/encoder/flac_encoder.cpp
View file

@ -85,7 +85,8 @@ void FlacEncoder::encode(const msg::PcmChunk& chunk)
pcmBuffer_ = static_cast<FLAC__int32*>(realloc(pcmBuffer_, pcmBufferSize_ * sizeof(FLAC__int32))); pcmBuffer_ = static_cast<FLAC__int32*>(realloc(pcmBuffer_, pcmBufferSize_ * sizeof(FLAC__int32)));
} }
auto clip = [](int32_t min, int32_t max, int32_t value) -> int32_t { auto clip = [](int32_t min, int32_t max, int32_t value) -> int32_t
{
if (value < min) if (value < min)
return min; return min;
if (value > max) if (value > max)

4
server/publishZeroConf/publish_avahi.cpp
View file

@ -71,7 +71,9 @@ void PublishAvahi::publish(const std::vector<mDNSService>& services)
void PublishAvahi::poll() void PublishAvahi::poll()
{ {
timer_.expires_after(std::chrono::milliseconds(50)); timer_.expires_after(std::chrono::milliseconds(50));
timer_.async_wait([this](const boost::system::error_code& ec) { timer_.async_wait(
[this](const boost::system::error_code& ec)
{
if (!ec && (avahi_simple_poll_iterate(simple_poll, 0) == 0)) if (!ec && (avahi_simple_poll_iterate(simple_poll, 0) == 0))
poll(); poll();
}); });

20
server/server.cpp
View file

@ -452,7 +452,8 @@ void Server::processRequest(const jsonrpcpp::request_ptr request, const OnRespon
auto command = request->params().get<string>("command"); auto command = request->params().get<string>("command");
auto handle_response = [request, on_response, command](const snapcast::ErrorCode& ec) { auto handle_response = [request, on_response, command](const snapcast::ErrorCode& ec)
{
auto log_level = AixLog::Severity::debug; auto log_level = AixLog::Severity::debug;
if (ec) if (ec)
log_level = AixLog::Severity::error; log_level = AixLog::Severity::error;
@ -533,7 +534,8 @@ void Server::processRequest(const jsonrpcpp::request_ptr request, const OnRespon
auto value = request->params().get("value"); auto value = request->params().get("value");
LOG(INFO, LOG_TAG) << "Stream '" << streamId << "' set property: " << name << " = " << value << "\n"; LOG(INFO, LOG_TAG) << "Stream '" << streamId << "' set property: " << name << " = " << value << "\n";
auto handle_response = [request, on_response](const snapcast::ErrorCode& ec) { auto handle_response = [request, on_response](const snapcast::ErrorCode& ec)
{
LOG(ERROR, LOG_TAG) << "SetShuffle: " << ec << ", message: " << ec.detailed_message() << ", msg: " << ec.message() LOG(ERROR, LOG_TAG) << "SetShuffle: " << ec << ", message: " << ec.detailed_message() << ", msg: " << ec.message()
<< ", category: " << ec.category().name() << "\n"; << ", category: " << ec.category().name() << "\n";
std::shared_ptr<jsonrpcpp::Response> response; std::shared_ptr<jsonrpcpp::Response> response;
@ -663,7 +665,9 @@ void Server::onMessageReceived(std::shared_ptr<ControlSession> controlSession, c
if (entity->is_request()) if (entity->is_request())
{ {
jsonrpcpp::request_ptr request = dynamic_pointer_cast<jsonrpcpp::Request>(entity); jsonrpcpp::request_ptr request = dynamic_pointer_cast<jsonrpcpp::Request>(entity);
processRequest(request, [this, controlSession, response_handler](jsonrpcpp::entity_ptr response, jsonrpcpp::notification_ptr notification) { processRequest(request,
[this, controlSession, response_handler](jsonrpcpp::entity_ptr response, jsonrpcpp::notification_ptr notification)
{
saveConfig(); saveConfig();
////cout << "Request: " << request->to_json().dump() << "\n"; ////cout << "Request: " << request->to_json().dump() << "\n";
if (notification) if (notification)
@ -693,8 +697,10 @@ void Server::onMessageReceived(std::shared_ptr<ControlSession> controlSession, c
if (batch_entity->is_request()) if (batch_entity->is_request())
{ {
jsonrpcpp::request_ptr request = dynamic_pointer_cast<jsonrpcpp::Request>(batch_entity); jsonrpcpp::request_ptr request = dynamic_pointer_cast<jsonrpcpp::Request>(batch_entity);
processRequest(request, [controlSession, response_handler, &responseBatch, &notificationBatch](jsonrpcpp::entity_ptr response, processRequest(request,
jsonrpcpp::notification_ptr notification) { [controlSession, response_handler, &responseBatch, &notificationBatch](jsonrpcpp::entity_ptr response,
jsonrpcpp::notification_ptr notification)
{
if (response != nullptr) if (response != nullptr)
responseBatch.add_ptr(response); responseBatch.add_ptr(response);
if (notification != nullptr) if (notification != nullptr)
@ -848,7 +854,9 @@ void Server::saveConfig(const std::chrono::milliseconds& deferred)
std::lock_guard<std::mutex> lock(mutex); std::lock_guard<std::mutex> lock(mutex);
config_timer_.cancel(); config_timer_.cancel();
config_timer_.expires_after(deferred); config_timer_.expires_after(deferred);
config_timer_.async_wait([](const boost::system::error_code& ec) { config_timer_.async_wait(
[](const boost::system::error_code& ec)
{
if (!ec) if (!ec)
{ {
LOG(DEBUG, LOG_TAG) << "Saving config\n"; LOG(DEBUG, LOG_TAG) << "Saving config\n";

4
server/snapserver.cpp
View file

@ -321,7 +321,9 @@ int main(int argc, char* argv[])
// Construct a signal set registered for process termination. // Construct a signal set registered for process termination.
boost::asio::signal_set signals(io_context, SIGHUP, SIGINT, SIGTERM); boost::asio::signal_set signals(io_context, SIGHUP, SIGINT, SIGTERM);
signals.async_wait([&io_context](const boost::system::error_code& ec, int signal) { signals.async_wait(
[&io_context](const boost::system::error_code& ec, int signal)
{
if (!ec) if (!ec)
LOG(INFO, LOG_TAG) << "Received signal " << signal << ": " << strsignal(signal) << "\n"; LOG(INFO, LOG_TAG) << "Received signal " << signal << ": " << strsignal(signal) << "\n";
else else

12
server/stream_server.cpp
View file

@ -142,10 +142,11 @@ void StreamServer::onDisconnect(StreamSession* streamSession)
LOG(INFO, LOG_TAG) << "onDisconnect: " << session->clientId << "\n"; LOG(INFO, LOG_TAG) << "onDisconnect: " << session->clientId << "\n";
LOG(DEBUG, LOG_TAG) << "sessions: " << sessions_.size() << "\n"; LOG(DEBUG, LOG_TAG) << "sessions: " << sessions_.size() << "\n";
sessions_.erase(std::remove_if(sessions_.begin(), sessions_.end(), sessions_.erase(std::remove_if(sessions_.begin(), sessions_.end(),
[streamSession](std::weak_ptr<StreamSession> session) { [streamSession](std::weak_ptr<StreamSession> session)
auto s = session.lock(); {
return s.get() == streamSession; auto s = session.lock();
}), return s.get() == streamSession;
}),
sessions_.end()); sessions_.end());
LOG(DEBUG, LOG_TAG) << "sessions: " << sessions_.size() << "\n"; LOG(DEBUG, LOG_TAG) << "sessions: " << sessions_.size() << "\n";
if (messageReceiver_ != nullptr) if (messageReceiver_ != nullptr)
@ -184,7 +185,8 @@ session_ptr StreamServer::getStreamSession(const std::string& clientId) const
void StreamServer::startAccept() void StreamServer::startAccept()
{ {
auto accept_handler = [this](error_code ec, tcp::socket socket) { auto accept_handler = [this](error_code ec, tcp::socket socket)
{
if (!ec) if (!ec)
handleAccept(std::move(socket)); handleAccept(std::move(socket));
else else

27
server/stream_session.cpp
View file

@ -60,8 +60,12 @@ void StreamSession::send_next()
{ {
auto& buffer = messages_.front(); auto& buffer = messages_.front();
buffer.on_air = true; buffer.on_air = true;
boost::asio::post(strand_, [this, self = shared_from_this(), buffer]() { boost::asio::post(strand_,
sendAsync(buffer, [this](boost::system::error_code ec, std::size_t length) { [this, self = shared_from_this(), buffer]()
{
sendAsync(buffer,
[this](boost::system::error_code ec, std::size_t length)
{
messages_.pop_front(); messages_.pop_front();
if (ec) if (ec)
{ {
@ -78,16 +82,19 @@ void StreamSession::send_next()
void StreamSession::send(shared_const_buffer const_buf) void StreamSession::send(shared_const_buffer const_buf)
{ {
boost::asio::post(strand_, [this, self = shared_from_this(), const_buf]() { boost::asio::post(strand_,
[this, self = shared_from_this(), const_buf]()
{
// delete PCM chunks that are older than the overall buffer duration // delete PCM chunks that are older than the overall buffer duration
messages_.erase(std::remove_if(messages_.begin(), messages_.end(), messages_.erase(std::remove_if(messages_.begin(), messages_.end(),
[this](const shared_const_buffer& buffer) { [this](const shared_const_buffer& buffer)
const auto& msg = buffer.message(); {
if (!msg.is_pcm_chunk || buffer.on_air) const auto& msg = buffer.message();
return false; if (!msg.is_pcm_chunk || buffer.on_air)
auto age = chronos::clk::now() - msg.rec_time; return false;
return (age > std::chrono::milliseconds(bufferMs_) + 100ms); auto age = chronos::clk::now() - msg.rec_time;
}), return (age > std::chrono::milliseconds(bufferMs_) + 100ms);
}),
messages_.end()); messages_.end());
messages_.push_back(const_buf); messages_.push_back(const_buf);

76
server/stream_session_tcp.cpp
View file

@ -86,50 +86,52 @@ std::string StreamSessionTcp::getIP()
void StreamSessionTcp::read_next() void StreamSessionTcp::read_next()
{ {
boost::asio::async_read( boost::asio::async_read(socket_, boost::asio::buffer(buffer_, base_msg_size_),
socket_, boost::asio::buffer(buffer_, base_msg_size_), [this, self = shared_from_this()](boost::system::error_code ec, std::size_t length) mutable { [this, self = shared_from_this()](boost::system::error_code ec, std::size_t length) mutable
{
if (ec)
{
LOG(ERROR, LOG_TAG) << "Error reading message header of length " << length << ": " << ec.message() << "\n";
messageReceiver_->onDisconnect(this);
return;
}
baseMessage_.deserialize(buffer_.data());
LOG(DEBUG, LOG_TAG) << "getNextMessage: " << baseMessage_.type << ", size: " << baseMessage_.size << ", id: " << baseMessage_.id
<< ", refers: " << baseMessage_.refersTo << "\n";
if (baseMessage_.type > message_type::kLast)
{
LOG(ERROR, LOG_TAG) << "unknown message type received: " << baseMessage_.type << ", size: " << baseMessage_.size << "\n";
messageReceiver_->onDisconnect(this);
return;
}
else if (baseMessage_.size > msg::max_size)
{
LOG(ERROR, LOG_TAG) << "received message of type " << baseMessage_.type << " to large: " << baseMessage_.size << "\n";
messageReceiver_->onDisconnect(this);
return;
}
if (baseMessage_.size > buffer_.size())
buffer_.resize(baseMessage_.size);
boost::asio::async_read(socket_, boost::asio::buffer(buffer_, baseMessage_.size),
[this, self](boost::system::error_code ec, std::size_t length) mutable
{
if (ec) if (ec)
{ {
LOG(ERROR, LOG_TAG) << "Error reading message header of length " << length << ": " << ec.message() << "\n"; LOG(ERROR, LOG_TAG) << "Error reading message body of length " << length << ": " << ec.message() << "\n";
messageReceiver_->onDisconnect(this); messageReceiver_->onDisconnect(this);
return; return;
} }
baseMessage_.deserialize(buffer_.data()); tv t;
LOG(DEBUG, LOG_TAG) << "getNextMessage: " << baseMessage_.type << ", size: " << baseMessage_.size << ", id: " << baseMessage_.id baseMessage_.received = t;
<< ", refers: " << baseMessage_.refersTo << "\n"; if (messageReceiver_ != nullptr)
if (baseMessage_.type > message_type::kLast) messageReceiver_->onMessageReceived(this, baseMessage_, buffer_.data());
{ read_next();
LOG(ERROR, LOG_TAG) << "unknown message type received: " << baseMessage_.type << ", size: " << baseMessage_.size << "\n";
messageReceiver_->onDisconnect(this);
return;
}
else if (baseMessage_.size > msg::max_size)
{
LOG(ERROR, LOG_TAG) << "received message of type " << baseMessage_.type << " to large: " << baseMessage_.size << "\n";
messageReceiver_->onDisconnect(this);
return;
}
if (baseMessage_.size > buffer_.size())
buffer_.resize(baseMessage_.size);
boost::asio::async_read(socket_, boost::asio::buffer(buffer_, baseMessage_.size),
[this, self](boost::system::error_code ec, std::size_t length) mutable {
if (ec)
{
LOG(ERROR, LOG_TAG) << "Error reading message body of length " << length << ": " << ec.message() << "\n";
messageReceiver_->onDisconnect(this);
return;
}
tv t;
baseMessage_.received = t;
if (messageReceiver_ != nullptr)
messageReceiver_->onMessageReceived(this, baseMessage_, buffer_.data());
read_next();
});
}); });
});
} }

4
server/streamreader/airplay_stream.cpp
View file

@ -245,7 +245,9 @@ void AirplayStream::pipeReadLine()
} }
const std::string delimiter = "\n"; const std::string delimiter = "\n";
boost::asio::async_read_until(*pipe_fd_, streambuf_pipe_, delimiter, [this, delimiter](const std::error_code& ec, std::size_t bytes_transferred) { boost::asio::async_read_until(*pipe_fd_, streambuf_pipe_, delimiter,
[this, delimiter](const std::error_code& ec, std::size_t bytes_transferred)
{
if (ec) if (ec)
{ {
if ((ec.value() == boost::asio::error::eof) || (ec.value() == boost::asio::error::bad_descriptor)) if ((ec.value() == boost::asio::error::eof) || (ec.value() == boost::asio::error::bad_descriptor))

4
server/streamreader/alsa_stream.cpp
View file

@ -54,7 +54,9 @@ template <typename Rep, typename Period>
void wait(boost::asio::steady_timer& timer, const std::chrono::duration<Rep, Period>& duration, std::function<void()> handler) void wait(boost::asio::steady_timer& timer, const std::chrono::duration<Rep, Period>& duration, std::function<void()> handler)
{ {
timer.expires_after(duration); timer.expires_after(duration);
timer.async_wait([handler = std::move(handler)](const boost::system::error_code& ec) { timer.async_wait(
[handler = std::move(handler)](const boost::system::error_code& ec)
{
if (ec) if (ec)
{ {
LOG(ERROR, LOG_TAG) << "Error during async wait: " << ec.message() << "\n"; LOG(ERROR, LOG_TAG) << "Error during async wait: " << ec.message() << "\n";

127
server/streamreader/asio_stream.hpp
View file

@ -76,7 +76,9 @@ template <typename Timer, typename Rep, typename Period>
void AsioStream<ReadStream>::wait(Timer& timer, const std::chrono::duration<Rep, Period>& duration, std::function<void()> handler) void AsioStream<ReadStream>::wait(Timer& timer, const std::chrono::duration<Rep, Period>& duration, std::function<void()> handler)
{ {
timer.expires_after(duration); timer.expires_after(duration);
timer.async_wait([handler = std::move(handler)](const boost::system::error_code& ec) { timer.async_wait(
[handler = std::move(handler)](const boost::system::error_code& ec)
{
if (ec) if (ec)
{ {
LOG(ERROR, "AsioStream") << "Error during async wait: " << ec.message() << "\n"; LOG(ERROR, "AsioStream") << "Error during async wait: " << ec.message() << "\n";
@ -114,7 +116,9 @@ template <typename ReadStream>
void AsioStream<ReadStream>::check_state() void AsioStream<ReadStream>::check_state()
{ {
uint64_t last_read = bytes_read_; uint64_t last_read = bytes_read_;
wait(state_timer_, std::chrono::milliseconds(500 + chunk_ms_), [this, last_read] { wait(state_timer_, std::chrono::milliseconds(500 + chunk_ms_),
[this, last_read]
{
LOG(TRACE, "AsioStream") << "check state last: " << last_read << ", read: " << bytes_read_ << "\n"; LOG(TRACE, "AsioStream") << "check state last: " << last_read << ", read: " << bytes_read_ << "\n";
if (bytes_read_ != last_read) if (bytes_read_ != last_read)
setState(ReaderState::kPlaying); setState(ReaderState::kPlaying);
@ -172,68 +176,71 @@ void AsioStream<ReadStream>::do_read()
{ {
// LOG(DEBUG, "AsioStream") << "do_read\n"; // LOG(DEBUG, "AsioStream") << "do_read\n";
boost::asio::async_read(*stream_, boost::asio::buffer(chunk_->payload, chunk_->payloadSize), boost::asio::async_read(*stream_, boost::asio::buffer(chunk_->payload, chunk_->payloadSize),
[this](boost::system::error_code ec, std::size_t length) mutable { [this](boost::system::error_code ec, std::size_t length) mutable
if (ec) {
{ if (ec)
LOG(ERROR, "AsioStream") << "Error reading message: " << ec.message() << ", length: " << length << "\n"; {
connect(); LOG(ERROR, "AsioStream") << "Error reading message: " << ec.message() << ", length: " << length << "\n";
return; connect();
} return;
}
bytes_read_ += length; bytes_read_ += length;
// LOG(DEBUG, "AsioStream") << "Read: " << length << " bytes\n"; // LOG(DEBUG, "AsioStream") << "Read: " << length << " bytes\n";
// First read after connect. Set the initial read timestamp // First read after connect. Set the initial read timestamp
// the timestamp will be incremented after encoding, // the timestamp will be incremented after encoding,
// since we do not know how much the encoder actually encoded // since we do not know how much the encoder actually encoded
if (!first_) if (!first_)
{ {
auto now = std::chrono::steady_clock::now(); auto now = std::chrono::steady_clock::now();
auto stream2systime_diff = now - tvEncodedChunk_; auto stream2systime_diff = now - tvEncodedChunk_;
if (stream2systime_diff > chronos::sec(5) + chronos::msec(chunk_ms_)) if (stream2systime_diff > chronos::sec(5) + chronos::msec(chunk_ms_))
{ {
LOG(WARNING, "AsioStream") << "Stream and system time out of sync: " LOG(WARNING, "AsioStream") << "Stream and system time out of sync: "
<< std::chrono::duration_cast<std::chrono::microseconds>(stream2systime_diff).count() / 1000. << std::chrono::duration_cast<std::chrono::microseconds>(stream2systime_diff).count() / 1000.
<< " ms, resetting stream time.\n"; << " ms, resetting stream time.\n";
first_ = true; first_ = true;
} }
} }
if (first_) if (first_)
{ {
first_ = false; first_ = false;
tvEncodedChunk_ = std::chrono::steady_clock::now() - chunk_->duration<std::chrono::nanoseconds>(); tvEncodedChunk_ = std::chrono::steady_clock::now() - chunk_->duration<std::chrono::nanoseconds>();
nextTick_ = std::chrono::steady_clock::now(); nextTick_ = std::chrono::steady_clock::now();
} }
chunkRead(*chunk_); chunkRead(*chunk_);
nextTick_ += chunk_->duration<std::chrono::nanoseconds>(); nextTick_ += chunk_->duration<std::chrono::nanoseconds>();
auto currentTick = std::chrono::steady_clock::now(); auto currentTick = std::chrono::steady_clock::now();
// Synchronize read to chunk_ms_ // Synchronize read to chunk_ms_
if (nextTick_ >= currentTick) if (nextTick_ >= currentTick)
{ {
read_timer_.expires_after(nextTick_ - currentTick); read_timer_.expires_after(nextTick_ - currentTick);
read_timer_.async_wait([this](const boost::system::error_code& ec) { read_timer_.async_wait(
if (ec) [this](const boost::system::error_code& ec)
{ {
LOG(ERROR, "AsioStream") << "Error during async wait: " << ec.message() << "\n"; if (ec)
} {
else LOG(ERROR, "AsioStream") << "Error during async wait: " << ec.message() << "\n";
{ }
do_read(); else
} {
}); do_read();
return; }
} });
// Read took longer, wait for the buffer to fill up return;
else }
{ // Read took longer, wait for the buffer to fill up
resync(std::chrono::duration_cast<std::chrono::nanoseconds>(currentTick - nextTick_)); else
nextTick_ = currentTick + std::chrono::milliseconds(buffer_ms_); {
first_ = true; resync(std::chrono::duration_cast<std::chrono::nanoseconds>(currentTick - nextTick_));
do_read(); nextTick_ = currentTick + std::chrono::milliseconds(buffer_ms_);
} first_ = true;
}); do_read();
}
});
} }
} // namespace streamreader } // namespace streamreader

3
server/streamreader/meta_stream.cpp
View file

@ -106,7 +106,8 @@ void MetaStream::onStateChanged(const PcmStream* pcmStream, ReaderState state)
// if (active_stream_->getProperties().playback_status == PlaybackStatus::kPaused) // if (active_stream_->getProperties().playback_status == PlaybackStatus::kPaused)
// return; // return;
auto switch_stream = [this](std::shared_ptr<PcmStream> new_stream) { auto switch_stream = [this](std::shared_ptr<PcmStream> new_stream)
{
if (new_stream == active_stream_) if (new_stream == active_stream_)
return; return;
LOG(INFO, LOG_TAG) << "Stream: " << name_ << ", switching active stream: " << (active_stream_ ? active_stream_->getName() : "<null>") << " => " LOG(INFO, LOG_TAG) << "Stream: " << name_ << ", switching active stream: " << (active_stream_ ? active_stream_->getName() : "<null>") << " => "

16
server/streamreader/pcm_stream.cpp
View file

@ -127,10 +127,14 @@ void PcmStream::onControlRequest(const jsonrpcpp::Request& request)
void PcmStream::pollProperties() void PcmStream::pollProperties()
{ {
property_timer_.expires_after(10s); property_timer_.expires_after(10s);
property_timer_.async_wait([this](const boost::system::error_code& ec) { property_timer_.async_wait(
[this](const boost::system::error_code& ec)
{
if (!ec) if (!ec)
{ {
stream_ctrl_->command({++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"; LOG(INFO, LOG_TAG) << "Response for Plugin.Stream.Player.GetProperties: " << response.to_json() << "\n";
if (response.error().code() == 0) if (response.error().code() == 0)
setProperties(response.result()); setProperties(response.result());
@ -154,7 +158,9 @@ void PcmStream::onControlNotification(const jsonrpcpp::Notification& notificatio
else if (notification.method() == "Plugin.Stream.Ready") else if (notification.method() == "Plugin.Stream.Ready")
{ {
LOG(DEBUG, LOG_TAG) << "Plugin is ready\n"; LOG(DEBUG, LOG_TAG) << "Plugin is ready\n";
stream_ctrl_->command({++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"; LOG(INFO, LOG_TAG) << "Response for Plugin.Stream.Player.GetProperties: " << response.to_json() << "\n";
if (response.error().code() == 0) if (response.error().code() == 0)
setProperties(response.result()); setProperties(response.result());
@ -453,7 +459,9 @@ void PcmStream::sendRequest(const std::string& method, const jsonrpcpp::Paramete
return handler({ControlErrc::can_not_control}); return handler({ControlErrc::can_not_control});
jsonrpcpp::Request req(++req_id_, method, params); jsonrpcpp::Request req(++req_id_, method, params);
stream_ctrl_->command(req, [handler](const jsonrpcpp::Response& response) { stream_ctrl_->command(req,
[handler](const jsonrpcpp::Response& response)
{
if (response.error().code() != 0) if (response.error().code() != 0)
handler({static_cast<ControlErrc>(response.error().code()), response.error().data()}); handler({static_cast<ControlErrc>(response.error().code()), response.error().data()});
else else

4
server/streamreader/process_stream.cpp
View file

@ -146,7 +146,9 @@ void ProcessStream::onStderrMsg(const std::string& line)
void ProcessStream::stderrReadLine() void ProcessStream::stderrReadLine()
{ {
const std::string delimiter = "\n"; 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) { boost::asio::async_read_until(*stream_stderr_, streambuf_stderr_, delimiter,
[this, delimiter](const std::error_code& ec, std::size_t bytes_transferred)
{
if (ec) if (ec)
{ {
LOG(ERROR, LOG_TAG) << "Error while reading from stderr: " << ec.message() << "\n"; LOG(ERROR, LOG_TAG) << "Error while reading from stderr: " << ec.message() << "\n";

24
server/streamreader/stream_control.cpp
View file

@ -67,7 +67,9 @@ void StreamControl::start(const std::string& stream_id, const ServerSettings& se
void StreamControl::command(const jsonrpcpp::Request& request, const OnResponse& response_handler) void StreamControl::command(const jsonrpcpp::Request& request, const OnResponse& response_handler)
{ {
// use strand to serialize commands sent from different threads // use strand to serialize commands sent from different threads
boost::asio::post(executor_, [this, request, response_handler]() { boost::asio::post(executor_,
[this, request, response_handler]()
{
if (response_handler) if (response_handler)
request_callbacks_[request.id()] = response_handler; request_callbacks_[request.id()] = response_handler;
@ -172,11 +174,13 @@ void ScriptStreamControl::doStart(const std::string& stream_id, const ServerSett
{ {
process_ = bp::child( process_ = bp::child(
script_ + params.str(), bp::std_out > pipe_stdout_, bp::std_err > pipe_stderr_, bp::std_in < in_, 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) { bp::on_exit =
auto severity = AixLog::Severity::debug; [](int exit, const std::error_code& ec_in)
if (exit != 0) {
severity = AixLog::Severity::error; auto severity = AixLog::Severity::debug;
LOG(severity, LOG_TAG) << "Exit code: " << exit << ", message: " << ec_in.message() << "\n"; if (exit != 0)
severity = AixLog::Severity::error;
LOG(severity, LOG_TAG) << "Exit code: " << exit << ", message: " << ec_in.message() << "\n";
}); });
} }
catch (const std::exception& e) catch (const std::exception& e)
@ -203,7 +207,9 @@ void ScriptStreamControl::doCommand(const jsonrpcpp::Request& request)
void ScriptStreamControl::stderrReadLine() void ScriptStreamControl::stderrReadLine()
{ {
const std::string delimiter = "\n"; 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) { boost::asio::async_read_until(*stream_stderr_, streambuf_stderr_, delimiter,
[this, delimiter](const std::error_code& ec, std::size_t bytes_transferred)
{
if (ec) if (ec)
{ {
LOG(ERROR, LOG_TAG) << "Error while reading from stderr: " << ec.message() << "\n"; LOG(ERROR, LOG_TAG) << "Error while reading from stderr: " << ec.message() << "\n";
@ -223,7 +229,9 @@ void ScriptStreamControl::stderrReadLine()
void ScriptStreamControl::stdoutReadLine() void ScriptStreamControl::stdoutReadLine()
{ {
const std::string delimiter = "\n"; 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) { boost::asio::async_read_until(*stream_stdout_, streambuf_stdout_, delimiter,
[this, delimiter](const std::error_code& ec, std::size_t bytes_transferred)
{
if (ec) if (ec)
{ {
LOG(ERROR, LOG_TAG) << "Error while reading from stdout: " << ec.message() << "\n"; LOG(ERROR, LOG_TAG) << "Error while reading from stdout: " << ec.message() << "\n";

8
server/streamreader/tcp_stream.cpp
View file

@ -75,7 +75,9 @@ void TcpStream::do_connect()
if (is_server_) if (is_server_)
{ {
acceptor_->async_accept([this](boost::system::error_code ec, tcp::socket socket) { acceptor_->async_accept(
[this](boost::system::error_code ec, tcp::socket socket)
{
if (!ec) if (!ec)
{ {
LOG(DEBUG, LOG_TAG) << "New client connection\n"; LOG(DEBUG, LOG_TAG) << "New client connection\n";
@ -92,7 +94,9 @@ void TcpStream::do_connect()
{ {
stream_ = make_unique<tcp::socket>(strand_); stream_ = make_unique<tcp::socket>(strand_);
boost::asio::ip::tcp::endpoint endpoint(boost::asio::ip::address::from_string(host_), port_); boost::asio::ip::tcp::endpoint endpoint(boost::asio::ip::address::from_string(host_), port_);
stream_->async_connect(endpoint, [this](const boost::system::error_code& ec) { stream_->async_connect(endpoint,
[this](const boost::system::error_code& ec)
{
if (!ec) if (!ec)
{ {
LOG(DEBUG, LOG_TAG) << "Connected\n"; LOG(DEBUG, LOG_TAG) << "Connected\n";

4
server/streamreader/watchdog.cpp
View file

@ -65,7 +65,9 @@ void Watchdog::trigger()
{ {
timer_.cancel(); timer_.cancel();
timer_.expires_after(timeout_ms_); timer_.expires_after(timeout_ms_);
timer_.async_wait([this](const boost::system::error_code& ec) { timer_.async_wait(
[this](const boost::system::error_code& ec)
{
if (!ec) if (!ec)
{ {
LOG(INFO, LOG_TAG) << "Timed out: " << std::chrono::duration_cast<std::chrono::seconds>(timeout_ms_).count() << "s\n"; LOG(INFO, LOG_TAG) << "Timed out: " << std::chrono::duration_cast<std::chrono::seconds>(timeout_ms_).count() << "s\n";