mirror of
https://github.com/badaix/snapcast
synced 2025-02-22 14:54:30 +01:00
122 lines
3.4 KiB
C++
122 lines
3.4 KiB
C++
/***
|
|
This file is part of snapcast
|
|
Copyright (C) 2014-2025 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/>.
|
|
***/
|
|
|
|
// prototype/interface header file
|
|
#include "stream_session.hpp"
|
|
|
|
// local headers
|
|
#include "common/aixlog.hpp"
|
|
|
|
// 3rd party headers
|
|
|
|
// standard headers
|
|
#include <iostream>
|
|
|
|
|
|
using namespace std;
|
|
using namespace streamreader;
|
|
|
|
static constexpr auto LOG_TAG = "StreamSession";
|
|
|
|
|
|
StreamSession::StreamSession(const boost::asio::any_io_executor& executor, StreamMessageReceiver* receiver)
|
|
: messageReceiver_(receiver), pcmStream_(nullptr), strand_(boost::asio::make_strand(executor))
|
|
{
|
|
base_msg_size_ = baseMessage_.getSize();
|
|
buffer_.resize(base_msg_size_);
|
|
}
|
|
|
|
|
|
void StreamSession::setPcmStream(PcmStreamPtr pcmStream)
|
|
{
|
|
std::lock_guard<std::mutex> lock(mutex_);
|
|
pcmStream_ = std::move(pcmStream);
|
|
}
|
|
|
|
|
|
const PcmStreamPtr StreamSession::pcmStream() const
|
|
{
|
|
std::lock_guard<std::mutex> lock(mutex_);
|
|
return pcmStream_;
|
|
}
|
|
|
|
|
|
void StreamSession::send_next()
|
|
{
|
|
auto& buffer = messages_.front();
|
|
buffer.on_air = true;
|
|
boost::asio::post(strand_, [this, self = shared_from_this(), buffer]()
|
|
{
|
|
sendAsync(buffer, [this](boost::system::error_code ec, std::size_t length)
|
|
{
|
|
messages_.pop_front();
|
|
if (ec)
|
|
{
|
|
LOG(ERROR, LOG_TAG) << "StreamSession write error (msg length: " << length << "): " << ec.message() << "\n";
|
|
messageReceiver_->onDisconnect(this);
|
|
return;
|
|
}
|
|
if (!messages_.empty())
|
|
send_next();
|
|
});
|
|
});
|
|
}
|
|
|
|
|
|
void StreamSession::send(shared_const_buffer const_buf)
|
|
{
|
|
boost::asio::post(strand_, [this, self = shared_from_this(), const_buf]()
|
|
{
|
|
// delete PCM chunks that are older than the overall buffer duration
|
|
messages_.erase(std::remove_if(messages_.begin(), messages_.end(),
|
|
[this](const shared_const_buffer& buffer)
|
|
{
|
|
const auto& msg = buffer.message();
|
|
if (!msg.is_pcm_chunk || buffer.on_air)
|
|
return false;
|
|
auto age = chronos::clk::now() - msg.rec_time;
|
|
return (age > std::chrono::milliseconds(bufferMs_) + 100ms);
|
|
}),
|
|
messages_.end());
|
|
|
|
messages_.push_back(const_buf);
|
|
|
|
if (messages_.size() > 1)
|
|
{
|
|
LOG(TRACE, LOG_TAG) << "outstanding async_write\n";
|
|
return;
|
|
}
|
|
send_next();
|
|
});
|
|
}
|
|
|
|
|
|
void StreamSession::send(msg::message_ptr message)
|
|
{
|
|
if (!message)
|
|
return;
|
|
|
|
// TODO: better set the timestamp in send_next for more accurate time sync
|
|
send(shared_const_buffer(*message));
|
|
}
|
|
|
|
|
|
void StreamSession::setBufferMs(size_t bufferMs)
|
|
{
|
|
bufferMs_ = bufferMs;
|
|
}
|