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

Add abstract AsioStream class

Browse source
This commit is contained in:
badaix 2019-11-27 20:44:34 +01:00
parent b00ee4aaf8
commit 48e76ea58d
5 changed files with 168 additions and 180 deletions
Download patch file Download diff file Expand all files Collapse all files

144
server/streamreader/asio_stream.hpp Normal file
View file

@ -0,0 +1,144 @@
/***
This file is part of snapcast
Copyright (C) 2014-2019 Johannes Pohl
This program is free software: you can redistribute it and/or modify
it under the terms of the GNU General Public License as published by
the Free Software Foundation, either version 3 of the License, or
(at your option) any later version.
This program is distributed in the hope that it will be useful,
but WITHOUT ANY WARRANTY; without even the implied warranty of
MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
GNU General Public License for more details.
You should have received a copy of the GNU General Public License
along with this program. If not, see <http://www.gnu.org/licenses/>.
***/
#ifndef ASIO_STREAM_HPP
#define ASIO_STREAM_HPP
#include "pcm_stream.hpp"
#include <boost/asio.hpp>
template <typename ReadStream>
class AsioStream : public PcmStream, public std::enable_shared_from_this<AsioStream<ReadStream>>
{
public:
/// ctor. Encoded PCM data is passed to the PipeListener
AsioStream(PcmListener* pcmListener, boost::asio::io_context& ioc, const StreamUri& uri);
void start() override;
void stop() override;
protected:
virtual void connect() = 0;
virtual void disconnect() = 0;
virtual void on_connect();
virtual void do_read();
std::unique_ptr<msg::PcmChunk> chunk_;
timeval tv_chunk_;
bool first_;
long nextTick_;
boost::asio::deadline_timer timer_;
std::unique_ptr<ReadStream> stream_;
};
template <typename ReadStream>
AsioStream<ReadStream>::AsioStream(PcmListener* pcmListener, boost::asio::io_context& ioc, const StreamUri& uri) : PcmStream(pcmListener, ioc, uri), timer_(ioc)
{
chunk_ = std::make_unique<msg::PcmChunk>(sampleFormat_, pcmReadMs_);
}
template <typename ReadStream>
void AsioStream<ReadStream>::start()
{
encoder_->init(this, sampleFormat_);
connect();
}
template <typename ReadStream>
void AsioStream<ReadStream>::stop()
{
timer_.cancel();
disconnect();
}
template <typename ReadStream>
void AsioStream<ReadStream>::on_connect()
{
chronos::systemtimeofday(&tv_chunk_);
tvEncodedChunk_ = tv_chunk_;
nextTick_ = chronos::getTickCount();
first_ = true;
do_read();
}
template <typename ReadStream>
void AsioStream<ReadStream>::do_read()
{
// LOG(DEBUG) << "do_read\n";
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),
[this, self](boost::system::error_code ec, std::size_t length) mutable {
if (ec)
{
LOG(ERROR) << "Error reading message: " << ec.message() << ", length: " << length << "\n";
connect();
return;
}
// LOG(DEBUG) << "Read: " << length << " bytes\n";
if (first_)
{
first_ = false;
chronos::systemtimeofday(&tv_chunk_);
chunk_->timestamp.sec = tv_chunk_.tv_sec;
chunk_->timestamp.usec = tv_chunk_.tv_usec;
tvEncodedChunk_ = tv_chunk_;
nextTick_ = chronos::getTickCount();
}
encoder_->encode(chunk_.get());
nextTick_ += pcmReadMs_;
chronos::addUs(tv_chunk_, pcmReadMs_ * 1000);
long currentTick = chronos::getTickCount();
if (nextTick_ >= currentTick)
{
setState(kPlaying);
timer_.expires_from_now(boost::posix_time::milliseconds(nextTick_ - currentTick));
timer_.async_wait([self, this](const boost::system::error_code& ec) {
if (ec)
{
LOG(ERROR) << "Error during async wait: " << ec.message() << "\n";
}
else
{
do_read();
}
});
return;
}
else
{
chronos::systemtimeofday(&tv_chunk_);
tvEncodedChunk_ = tv_chunk_;
pcmListener_->onResync(this, currentTick - nextTick_);
nextTick_ = currentTick;
do_read();
}
});
}
#endif

86
server/streamreader/pipe_stream.cpp
View file

@ -33,7 +33,8 @@ using namespace std;
PipeStream::PipeStream(PcmListener* pcmListener, boost::asio::io_context& ioc, const StreamUri& uri) : PcmStream(pcmListener, ioc, uri), timer_(ioc)
PipeStream::PipeStream(PcmListener* pcmListener, boost::asio::io_context& ioc, const StreamUri& uri)
: AsioStream<boost::asio::posix::stream_descriptor>(pcmListener, ioc, uri)
{
umask(0);
string mode = uri_.getQuery("mode", "create");
@ -51,91 +52,20 @@ PipeStream::PipeStream(PcmListener* pcmListener, boost::asio::io_context& ioc, c
}
PipeStream::~PipeStream()
void PipeStream::connect()
{
fifo_->close();
}
void PipeStream::start()
{
encoder_->init(this, sampleFormat_);
active_ = true;
do_accept();
}
void PipeStream::do_accept()
{
LOG(DEBUG) << "do_accept\n";
LOG(DEBUG) << "connect\n";
auto self = shared_from_this();
auto fd = open(uri_.path.c_str(), O_RDONLY | O_NONBLOCK);
fifo_ = std::make_unique<boost::asio::posix::stream_descriptor>(ioc_, fd);
chronos::systemtimeofday(&tv_chunk_);
tvEncodedChunk_ = tv_chunk_;
nextTick_ = chronos::getTickCount();
first_ = true;
do_read();
stream_ = std::make_unique<boost::asio::posix::stream_descriptor>(ioc_, fd);
on_connect();
}
void PipeStream::do_read()
void PipeStream::disconnect()
{
// LOG(DEBUG) << "do_read\n";
auto self = shared_from_this();
chunk_->timestamp.sec = tv_chunk_.tv_sec;
chunk_->timestamp.usec = tv_chunk_.tv_usec;
boost::asio::async_read(*fifo_, boost::asio::buffer(chunk_->payload, chunk_->payloadSize),
[this, self](boost::system::error_code ec, std::size_t length) mutable {
if (ec)
{
LOG(ERROR) << "Error reading message: " << ec.message() << ", length: " << length << "\n";
do_accept();
return;
stream_->close();
}
// LOG(DEBUG) << "Read: " << length << " bytes\n";
if (first_)
{
first_ = false;
chronos::systemtimeofday(&tv_chunk_);
chunk_->timestamp.sec = tv_chunk_.tv_sec;
chunk_->timestamp.usec = tv_chunk_.tv_usec;
tvEncodedChunk_ = tv_chunk_;
nextTick_ = chronos::getTickCount();
}
encoder_->encode(chunk_.get());
nextTick_ += pcmReadMs_;
chronos::addUs(tv_chunk_, pcmReadMs_ * 1000);
long currentTick = chronos::getTickCount();
if (nextTick_ >= currentTick)
{
setState(kPlaying);
timer_.expires_from_now(boost::posix_time::milliseconds(nextTick_ - currentTick));
timer_.async_wait([self, this](const boost::system::error_code& ec) {
if (ec)
{
LOG(ERROR) << "Error during async wait: " << ec.message() << "\n";
}
else
{
do_read();
}
});
return;
}
else
{
chronos::systemtimeofday(&tv_chunk_);
tvEncodedChunk_ = tv_chunk_;
pcmListener_->onResync(this, currentTick - nextTick_);
nextTick_ = currentTick;
do_read();
}
});
}
// void PipeStream::worker()
// {

20
server/streamreader/pipe_stream.hpp
View file

@ -19,8 +19,9 @@
#ifndef PIPE_STREAM_H
#define PIPE_STREAM_H
#include "pcm_stream.hpp"
#include <boost/asio.hpp>
#include "asio_stream.hpp"
using boost::asio::posix::stream_descriptor;
/// Reads and decodes PCM data from a named pipe
@ -29,24 +30,15 @@
* Implements EncoderListener to get the encoded data.
* Data is passed to the PcmListener
*/
class PipeStream : public PcmStream, public std::enable_shared_from_this<PipeStream>
class PipeStream : public AsioStream<stream_descriptor>
{
public:
/// ctor. Encoded PCM data is passed to the PipeListener
PipeStream(PcmListener* pcmListener, boost::asio::io_context& ioc, const StreamUri& uri);
~PipeStream() override;
void start() override;
protected:
void do_accept();
void do_read();
std::unique_ptr<msg::PcmChunk> chunk_;
timeval tv_chunk_;
bool first_;
long nextTick_;
boost::asio::deadline_timer timer_;
std::unique_ptr<boost::asio::posix::stream_descriptor> fifo_;
void connect() override;
void disconnect() override;
};

79
server/streamreader/tcp_stream.cpp
View file

@ -33,7 +33,7 @@ using namespace std;
TcpStream::TcpStream(PcmListener* pcmListener, boost::asio::io_context& ioc, const StreamUri& uri) : PcmStream(pcmListener, ioc, uri), timer_(ioc)
TcpStream::TcpStream(PcmListener* pcmListener, boost::asio::io_context& ioc, const StreamUri& uri) : AsioStream<tcp::socket>(pcmListener, ioc, uri)
{
size_t port = 4953;
try
@ -46,35 +46,18 @@ TcpStream::TcpStream(PcmListener* pcmListener, boost::asio::io_context& ioc, con
LOG(INFO) << "TcpStream port: " << port << "\n";
acceptor_ = make_unique<tcp::acceptor>(ioc_, tcp::endpoint(tcp::v4(), port));
chunk_ = make_unique<msg::PcmChunk>(sampleFormat_, pcmReadMs_);
LOG(DEBUG) << "Chunk size: " << chunk_->payloadSize << "\n";
}
TcpStream::~TcpStream()
{
}
void TcpStream::start()
{
encoder_->init(this, sampleFormat_);
active_ = true;
do_accept();
}
void TcpStream::do_accept()
void TcpStream::connect()
{
auto self = shared_from_this();
acceptor_->async_accept([this, self](boost::system::error_code ec, tcp::socket socket) {
if (!ec)
{
LOG(DEBUG) << "New client connection\n";
socket_ = make_unique<tcp::socket>(move(socket));
tv_chunk_.tv_sec = 0;
first_ = true;
do_read();
stream_ = make_unique<tcp::socket>(move(socket));
on_connect();
}
else
{
@ -84,57 +67,7 @@ void TcpStream::do_accept()
}
void TcpStream::do_read()
void TcpStream::disconnect()
{
auto self = shared_from_this();
chunk_->timestamp.sec = tv_chunk_.tv_sec;
chunk_->timestamp.usec = tv_chunk_.tv_usec;
boost::asio::async_read(*socket_, boost::asio::buffer(chunk_->payload, chunk_->payloadSize),
[this, self](boost::system::error_code ec, std::size_t length) mutable {
if (ec)
{
LOG(ERROR) << "Error reading message: " << ec.message() << "\n";
do_accept();
return;
}
LOG(DEBUG) << "Read: " << length << " bytes\n";
if (first_)
{
first_ = false;
chronos::systemtimeofday(&tv_chunk_);
chunk_->timestamp.sec = tv_chunk_.tv_sec;
chunk_->timestamp.usec = tv_chunk_.tv_usec;
tvEncodedChunk_ = tv_chunk_;
nextTick_ = chronos::getTickCount();
}
encoder_->encode(chunk_.get());
nextTick_ += pcmReadMs_;
chronos::addUs(tv_chunk_, pcmReadMs_ * 1000);
long currentTick = chronos::getTickCount();
if (nextTick_ >= currentTick)
{
setState(kPlaying);
timer_.expires_from_now(boost::posix_time::milliseconds(nextTick_ - currentTick));
timer_.async_wait([self, this](const boost::system::error_code& ec) {
if (ec)
{
LOG(ERROR) << "Error during async wait: " << ec.message() << "\n";
}
else
{
do_read();
}
});
return;
}
else
{
chronos::systemtimeofday(&tv_chunk_);
tvEncodedChunk_ = tv_chunk_;
pcmListener_->onResync(this, currentTick - nextTick_);
nextTick_ = currentTick;
do_read();
}
});
stream_->close();
}

19
server/streamreader/tcp_stream.hpp
View file

@ -19,9 +19,7 @@
#ifndef TCP_STREAM_H
#define TCP_STREAM_H
#include "pcm_stream.hpp"
#include <boost/asio.hpp>
#include <memory>
#include "asio_stream.hpp"
using boost::asio::ip::tcp;
@ -32,25 +30,16 @@ using boost::asio::ip::tcp;
* Implements EncoderListener to get the encoded data.
* Data is passed to the PcmListener
*/
class TcpStream : public PcmStream, public std::enable_shared_from_this<TcpStream>
class TcpStream : public AsioStream<tcp::socket>
{
public:
/// ctor. Encoded PCM data is passed to the PipeListener
TcpStream(PcmListener* pcmListener, boost::asio::io_context& ioc, const StreamUri& uri);
~TcpStream() override;
void start() override;
protected:
void do_accept();
void do_read();
void connect() override;
void disconnect() override;
std::unique_ptr<tcp::acceptor> acceptor_;
std::unique_ptr<tcp::socket> socket_;
std::unique_ptr<msg::PcmChunk> chunk_;
timeval tv_chunk_;
bool first_;
long nextTick_;
boost::asio::deadline_timer timer_;
};