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

Clean up AsioStream reader code

Browse source
This commit is contained in:
badaix 2019-12-01 11:38:55 +01:00
parent 3cb9902bbe
commit a47849a571
8 changed files with 35 additions and 137 deletions
Download patch file Download diff file Expand all files Collapse all files

2
server/stream_server.cpp
View file

@ -80,7 +80,7 @@ void StreamServer::onStateChanged(const PcmStream* pcmStream, const ReaderState&
// clang-format off // clang-format off
// Notification: {"jsonrpc":"2.0","method":"Stream.OnUpdate","params":{"id":"stream 1","stream":{"id":"stream 1","status":"idle","uri":{"fragment":"","host":"","path":"/tmp/snapfifo","query":{"buffer_ms":"20","codec":"flac","name":"stream 1","sampleformat":"48000:16:2"},"raw":"pipe:///tmp/snapfifo?name=stream 1","scheme":"pipe"}}}} // Notification: {"jsonrpc":"2.0","method":"Stream.OnUpdate","params":{"id":"stream 1","stream":{"id":"stream 1","status":"idle","uri":{"fragment":"","host":"","path":"/tmp/snapfifo","query":{"buffer_ms":"20","codec":"flac","name":"stream 1","sampleformat":"48000:16:2"},"raw":"pipe:///tmp/snapfifo?name=stream 1","scheme":"pipe"}}}}
// clang-format on // clang-format on
LOG(INFO) << "onStateChanged (" << pcmStream->getName() << "): " << state << "\n"; LOG(INFO) << "onStateChanged (" << pcmStream->getName() << "): " << static_cast<int>(state) << "\n";
// LOG(INFO) << pcmStream->toJson().dump(4); // LOG(INFO) << pcmStream->toJson().dump(4);
json notification = jsonrpcpp::Notification("Stream.OnUpdate", jsonrpcpp::Parameter("id", pcmStream->getId(), "stream", pcmStream->toJson())).to_json(); json notification = jsonrpcpp::Notification("Stream.OnUpdate", jsonrpcpp::Parameter("id", pcmStream->getId(), "stream", pcmStream->toJson())).to_json();
controlServer_->send(notification.dump(), nullptr); controlServer_->send(notification.dump(), nullptr);

35
server/streamreader/asio_stream.hpp
View file

@ -46,8 +46,8 @@ protected:
bool first_; bool first_;
long nextTick_; long nextTick_;
uint32_t buffer_ms_; uint32_t buffer_ms_;
boost::asio::deadline_timer timer_; boost::asio::deadline_timer read_timer_;
boost::asio::deadline_timer idle_timer_; boost::asio::deadline_timer state_timer_;
std::unique_ptr<ReadStream> stream_; std::unique_ptr<ReadStream> stream_;
std::atomic<std::uint64_t> bytes_read_; std::atomic<std::uint64_t> bytes_read_;
}; };
@ -56,7 +56,7 @@ protected:
template <typename ReadStream> template <typename ReadStream>
AsioStream<ReadStream>::AsioStream(PcmListener* pcmListener, boost::asio::io_context& ioc, const StreamUri& uri) AsioStream<ReadStream>::AsioStream(PcmListener* pcmListener, boost::asio::io_context& ioc, const StreamUri& uri)
: PcmStream(pcmListener, ioc, uri), timer_(ioc), idle_timer_(ioc) : PcmStream(pcmListener, ioc, uri), read_timer_(ioc), state_timer_(ioc)
{ {
chunk_ = std::make_unique<msg::PcmChunk>(sampleFormat_, pcmReadMs_); chunk_ = std::make_unique<msg::PcmChunk>(sampleFormat_, pcmReadMs_);
bytes_read_ = 0; bytes_read_ = 0;
@ -77,15 +77,15 @@ void AsioStream<ReadStream>::check_state()
{ {
uint64_t last_read = bytes_read_; uint64_t last_read = bytes_read_;
auto self = this->shared_from_this(); auto self = this->shared_from_this();
idle_timer_.expires_from_now(boost::posix_time::milliseconds(500 + pcmReadMs_)); state_timer_.expires_from_now(boost::posix_time::milliseconds(500 + pcmReadMs_));
idle_timer_.async_wait([self, this, last_read](const boost::system::error_code& ec) { state_timer_.async_wait([self, this, last_read](const boost::system::error_code& ec) {
if (!ec) if (!ec)
{ {
LOG(DEBUG) << "check state last: " << last_read << ", read: " << bytes_read_ << "\n"; LOG(DEBUG) << "check state last: " << last_read << ", read: " << bytes_read_ << "\n";
if (bytes_read_ != last_read) if (bytes_read_ != last_read)
setState(kPlaying); setState(ReaderState::kPlaying);
else else
setState(kIdle); setState(ReaderState::kIdle);
check_state(); check_state();
} }
}); });
@ -106,8 +106,8 @@ template <typename ReadStream>
void AsioStream<ReadStream>::stop() void AsioStream<ReadStream>::stop()
{ {
active_ = false; active_ = false;
timer_.cancel(); read_timer_.cancel();
idle_timer_.cancel(); state_timer_.cancel();
disconnect(); disconnect();
} }
@ -115,9 +115,6 @@ void AsioStream<ReadStream>::stop()
template <typename ReadStream> template <typename ReadStream>
void AsioStream<ReadStream>::on_connect() void AsioStream<ReadStream>::on_connect()
{ {
chronos::systemtimeofday(&tv_chunk_);
tvEncodedChunk_ = tv_chunk_;
nextTick_ = chronos::getTickCount();
first_ = true; first_ = true;
do_read(); do_read();
} }
@ -128,8 +125,6 @@ void AsioStream<ReadStream>::do_read()
{ {
// LOG(DEBUG) << "do_read\n"; // LOG(DEBUG) << "do_read\n";
auto self = this->shared_from_this(); auto self = this->shared_from_this();
chunk_->timestamp.sec = tv_chunk_.tv_sec;
chunk_->timestamp.usec = tv_chunk_.tv_usec;
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, self](boost::system::error_code ec, std::size_t length) mutable { [this, self](boost::system::error_code ec, std::size_t length) mutable {
if (ec) if (ec)
@ -144,21 +139,17 @@ void AsioStream<ReadStream>::do_read()
if (first_) if (first_)
{ {
first_ = false; first_ = false;
chronos::systemtimeofday(&tv_chunk_); chronos::systemtimeofday(&tvEncodedChunk_);
chunk_->timestamp.sec = tv_chunk_.tv_sec;
chunk_->timestamp.usec = tv_chunk_.tv_usec;
tvEncodedChunk_ = tv_chunk_;
nextTick_ = chronos::getTickCount() + buffer_ms_; nextTick_ = chronos::getTickCount() + buffer_ms_;
} }
encoder_->encode(chunk_.get()); encoder_->encode(chunk_.get());
nextTick_ += pcmReadMs_; nextTick_ += pcmReadMs_;
chronos::addUs(tv_chunk_, pcmReadMs_ * 1000);
long currentTick = chronos::getTickCount(); long currentTick = chronos::getTickCount();
if (nextTick_ >= currentTick) if (nextTick_ >= currentTick)
{ {
timer_.expires_from_now(boost::posix_time::milliseconds(nextTick_ - currentTick)); read_timer_.expires_from_now(boost::posix_time::milliseconds(nextTick_ - currentTick));
timer_.async_wait([self, this](const boost::system::error_code& ec) { read_timer_.async_wait([self, this](const boost::system::error_code& ec) {
if (ec) if (ec)
{ {
LOG(ERROR) << "Error during async wait: " << ec.message() << "\n"; LOG(ERROR) << "Error during async wait: " << ec.message() << "\n";
@ -172,8 +163,6 @@ void AsioStream<ReadStream>::do_read()
} }
else else
{ {
chronos::systemtimeofday(&tv_chunk_);
tvEncodedChunk_ = tv_chunk_;
pcmListener_->onResync(this, currentTick - nextTick_); pcmListener_->onResync(this, currentTick - nextTick_);
nextTick_ = currentTick + buffer_ms_; nextTick_ = currentTick + buffer_ms_;
do_read(); do_read();

2
server/streamreader/file_stream.cpp
View file

@ -56,7 +56,7 @@ void FileStream::worker()
size_t length = ifs.tellg(); size_t length = ifs.tellg();
ifs.seekg(0, ifs.beg); ifs.seekg(0, ifs.beg);
setState(kPlaying); setState(ReaderState::kPlaying);
while (active_) while (active_)
{ {

21
server/streamreader/pcm_stream.cpp
View file

@ -32,7 +32,7 @@ using namespace std;
PcmStream::PcmStream(PcmListener* pcmListener, boost::asio::io_context& ioc, const StreamUri& uri) PcmStream::PcmStream(PcmListener* pcmListener, boost::asio::io_context& ioc, const StreamUri& uri)
: active_(false), pcmListener_(pcmListener), uri_(uri), pcmReadMs_(20), state_(kIdle), ioc_(ioc) : active_(false), pcmListener_(pcmListener), uri_(uri), pcmReadMs_(20), state_(ReaderState::kIdle), ioc_(ioc)
{ {
encoder::EncoderFactory encoderFactory; encoder::EncoderFactory encoderFactory;
if (uri_.query.find("codec") == uri_.query.end()) if (uri_.query.find("codec") == uri_.query.end())
@ -51,11 +51,6 @@ PcmStream::PcmStream(PcmListener* pcmListener, boost::asio::io_context& ioc, con
if (uri_.query.find("buffer_ms") != uri_.query.end()) if (uri_.query.find("buffer_ms") != uri_.query.end())
pcmReadMs_ = cpt::stoul(uri_.query["buffer_ms"]); pcmReadMs_ = cpt::stoul(uri_.query["buffer_ms"]);
if (uri_.query.find("dryout_ms") != uri_.query.end())
dryoutMs_ = cpt::stoul(uri_.query["dryout_ms"]);
else
dryoutMs_ = 2000;
// meta_.reset(new msg::StreamTags()); // meta_.reset(new msg::StreamTags());
// meta_->msg["stream"] = name_; // meta_->msg["stream"] = name_;
setMeta(json()); setMeta(json());
@ -138,7 +133,7 @@ void PcmStream::setState(const ReaderState& newState)
{ {
if (newState != state_) if (newState != state_)
{ {
LOG(DEBUG) << "State changed: " << state_ << " => " << newState << "\n"; LOG(DEBUG) << "State changed: " << static_cast<int>(state_) << " => " << static_cast<int>(newState) << "\n";
state_ = newState; state_ = newState;
if (pcmListener_) if (pcmListener_)
pcmListener_->onStateChanged(this, newState); pcmListener_->onStateChanged(this, newState);
@ -163,15 +158,17 @@ void PcmStream::onChunkEncoded(const encoder::Encoder* /*encoder*/, msg::PcmChun
json PcmStream::toJson() const json PcmStream::toJson() const
{ {
string state("unknown"); string state("unknown");
if (state_ == kIdle) if (state_ == ReaderState::kIdle)
state = "idle"; state = "idle";
else if (state_ == kPlaying) else if (state_ == ReaderState::kPlaying)
state = "playing"; state = "playing";
else if (state_ == kDisabled) else if (state_ == ReaderState::kDisabled)
state = "disabled"; state = "disabled";
json j = { json j = {
{"uri", uri_.toJson()}, {"id", getId()}, {"status", state}, {"uri", uri_.toJson()},
{"id", getId()},
{"status", state},
}; };
if (meta_) if (meta_)
@ -185,7 +182,7 @@ std::shared_ptr<msg::StreamTags> PcmStream::getMeta() const
return meta_; return meta_;
} }
void PcmStream::setMeta(json jtag) void PcmStream::setMeta(const json& jtag)
{ {
meta_.reset(new msg::StreamTags(jtag)); meta_.reset(new msg::StreamTags(jtag));
meta_->msg["STREAM"] = name_; meta_->msg["STREAM"] = name_;

5
server/streamreader/pcm_stream.hpp
View file

@ -36,7 +36,7 @@
class PcmStream; class PcmStream;
enum ReaderState enum class ReaderState
{ {
kUnknown = 0, kUnknown = 0,
kIdle = 1, kIdle = 1,
@ -85,7 +85,7 @@ public:
virtual const SampleFormat& getSampleFormat() const; virtual const SampleFormat& getSampleFormat() const;
std::shared_ptr<msg::StreamTags> getMeta() const; std::shared_ptr<msg::StreamTags> getMeta() const;
void setMeta(json j); void setMeta(const json& j);
virtual ReaderState getState() const; virtual ReaderState getState() const;
virtual json toJson() const; virtual json toJson() const;
@ -106,7 +106,6 @@ protected:
StreamUri uri_; StreamUri uri_;
SampleFormat sampleFormat_; SampleFormat sampleFormat_;
size_t pcmReadMs_; size_t pcmReadMs_;
size_t dryoutMs_;
std::unique_ptr<encoder::Encoder> encoder_; std::unique_ptr<encoder::Encoder> encoder_;
std::string name_; std::string name_;
ReaderState state_; ReaderState state_;

94
server/streamreader/pipe_stream.cpp
View file

@ -65,97 +65,3 @@ void PipeStream::disconnect()
stream_->close(); stream_->close();
} }
// void PipeStream::worker()
// {
// timeval tvChunk;
// std::unique_ptr<msg::PcmChunk> chunk(new msg::PcmChunk(sampleFormat_, pcmReadMs_));
// string lastException = "";
// while (active_)
// {
// if (fd_ != -1)
// close(fd_);
// fd_ = open(uri_.path.c_str(), O_RDONLY | O_NONBLOCK);
// chronos::systemtimeofday(&tvChunk);
// tvEncodedChunk_ = tvChunk;
// long nextTick = chronos::getTickCount();
// int idleBytes = 0;
// int maxIdleBytes = sampleFormat_.rate * sampleFormat_.frameSize * dryoutMs_ / 1000;
// try
// {
// if (fd_ == -1)
// throw SnapException("failed to open fifo: \"" + uri_.path + "\"");
// while (active_)
// {
// chunk->timestamp.sec = tvChunk.tv_sec;
// chunk->timestamp.usec = tvChunk.tv_usec;
// int toRead = chunk->payloadSize;
// int len = 0;
// do
// {
// int count = read(fd_, chunk->payload + len, toRead - len);
// if (count < 0 && idleBytes < maxIdleBytes)
// {
// memset(chunk->payload + len, 0, toRead - len);
// idleBytes += toRead - len;
// len += toRead - len;
// continue;
// }
// if (count < 0)
// {
// setState(kIdle);
// if (!sleep(100))
// break;
// }
// else if (count == 0)
// throw SnapException("end of file");
// else
// {
// len += count;
// idleBytes = 0;
// }
// } while ((len < toRead) && active_);
// if (!active_)
// break;
// /// TODO: use less raw pointers, make this encoding more transparent
// encoder_->encode(chunk.get());
// if (!active_)
// break;
// nextTick += pcmReadMs_;
// chronos::addUs(tvChunk, pcmReadMs_ * 1000);
// long currentTick = chronos::getTickCount();
// if (nextTick >= currentTick)
// {
// setState(kPlaying);
// if (!sleep(nextTick - currentTick))
// break;
// }
// else
// {
// chronos::systemtimeofday(&tvChunk);
// tvEncodedChunk_ = tvChunk;
// pcmListener_->onResync(this, currentTick - nextTick);
// nextTick = currentTick;
// }
// lastException = "";
// }
// }
// catch (const std::exception& e)
// {
// if (lastException != e.what())
// {
// LOG(ERROR) << "(PipeStream) Exception: " << e.what() << std::endl;
// lastException = e.what();
// }
// if (!sleep(100))
// break;
// }
// }
// }

11
server/streamreader/process_stream.cpp
View file

@ -36,6 +36,11 @@ ProcessStream::ProcessStream(PcmListener* pcmListener, boost::asio::io_context&
{ {
params_ = uri_.getQuery("params"); params_ = uri_.getQuery("params");
logStderr_ = (uri_.getQuery("logStderr", "false") == "true"); logStderr_ = (uri_.getQuery("logStderr", "false") == "true");
if (uri_.query.find("dryout_ms") != uri_.query.end())
dryoutMs_ = cpt::stoul(uri_.query["dryout_ms"]);
else
dryoutMs_ = 2000;
} }
@ -143,7 +148,7 @@ void ProcessStream::worker()
{ {
timeval tvChunk; timeval tvChunk;
std::unique_ptr<msg::PcmChunk> chunk(new msg::PcmChunk(sampleFormat_, pcmReadMs_)); std::unique_ptr<msg::PcmChunk> chunk(new msg::PcmChunk(sampleFormat_, pcmReadMs_));
setState(kPlaying); setState(ReaderState::kPlaying);
string lastException = ""; string lastException = "";
while (active_) while (active_)
@ -180,7 +185,7 @@ void ProcessStream::worker()
} }
if (count < 0) if (count < 0)
{ {
setState(kIdle); setState(ReaderState::kIdle);
if (!sleep(100)) if (!sleep(100))
break; break;
} }
@ -207,7 +212,7 @@ void ProcessStream::worker()
if (nextTick >= currentTick) if (nextTick >= currentTick)
{ {
setState(kPlaying); setState(ReaderState::kPlaying);
if (!sleep(nextTick - currentTick)) if (!sleep(nextTick - currentTick))
break; break;
} }

2
server/streamreader/process_stream.hpp
View file

@ -25,6 +25,7 @@
#include "pcm_stream.hpp" #include "pcm_stream.hpp"
#include "process.hpp" #include "process.hpp"
// TODO: switch to AsioStream, maybe use boost::process library
/// Starts an external process and reads and PCM data from stdout /// Starts an external process and reads and PCM data from stdout
/** /**
@ -49,6 +50,7 @@ protected:
std::unique_ptr<Process> process_; std::unique_ptr<Process> process_;
std::thread stderrReaderThread_; std::thread stderrReaderThread_;
bool logStderr_; bool logStderr_;
size_t dryoutMs_;
void worker() override; void worker() override;
virtual void stderrReader(); virtual void stderrReader();