3pp-mirror/snapcast
1
0
Fork 0
mirror of https://github.com/badaix/snapcast.git synced 2025-04-29 18:27:12 +02:00
Code Issues Projects Releases Packages Wiki Activity

code cleanup

Browse source 
git-svn-id: svn://elaine/murooma/trunk@337 d8a302eb-03bc-478d-80e4-98257eca68ef
This commit is contained in:
(no author) 2015-01-01 18:15:20 +00:00
parent 2ea718dee0
commit 0a20924e66
15 changed files with 332 additions and 325 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

6
Makefile
View file

@ -18,4 +18,10 @@ installclient:
installserver: installserver:
$(MAKE) install -C server $(MAKE) install -C server
uninstallclient:
$(MAKE) uninstall -C client
uninstallserver:
$(MAKE) uninstall -C server

81
client/alsaPlayer.cpp
View file

@ -7,7 +7,7 @@
using namespace std; using namespace std;
Player::Player(const PcmDevice& pcmDevice, Stream* stream) : pcm_handle(NULL), buff(NULL), active_(false), stream_(stream), pcmDevice_(pcmDevice) Player::Player(const PcmDevice& pcmDevice, Stream* stream) : pcm_handle_(NULL), buff_(NULL), active_(false), stream_(stream), pcmDevice_(pcmDevice)
{ {
} }
@ -19,11 +19,12 @@ void Player::start()
snd_pcm_hw_params_t *params; snd_pcm_hw_params_t *params;
int buff_size; int buff_size;
rate = stream_->format.rate; const msg::SampleFormat& format = stream_->getFormat();
channels = stream_->format.channels; rate = format.rate;
channels = format.channels;
/* Open the PCM device in playback mode */ /* Open the PCM device in playback mode */
if ((pcm = snd_pcm_open(&pcm_handle, pcmDevice_.name.c_str(), SND_PCM_STREAM_PLAYBACK, 0)) < 0) if ((pcm = snd_pcm_open(&pcm_handle_, pcmDevice_.name.c_str(), SND_PCM_STREAM_PLAYBACK, 0)) < 0)
logE << "ERROR: Can't open " << pcmDevice_.name << " PCM device. " << snd_strerror(pcm) << "\n"; logE << "ERROR: Can't open " << pcmDevice_.name << " PCM device. " << snd_strerror(pcm) << "\n";
/* struct snd_pcm_playback_info_t pinfo; /* struct snd_pcm_playback_info_t pinfo;
@ -34,19 +35,19 @@ void Player::start()
/* Allocate parameters object and fill it with default values*/ /* Allocate parameters object and fill it with default values*/
snd_pcm_hw_params_alloca(&params); snd_pcm_hw_params_alloca(&params);
snd_pcm_hw_params_any(pcm_handle, params); snd_pcm_hw_params_any(pcm_handle_, params);
/* Set parameters */ /* Set parameters */
if ((pcm = snd_pcm_hw_params_set_access(pcm_handle, params, SND_PCM_ACCESS_RW_INTERLEAVED)) < 0) if ((pcm = snd_pcm_hw_params_set_access(pcm_handle_, params, SND_PCM_ACCESS_RW_INTERLEAVED)) < 0)
logE << "ERROR: Can't set interleaved mode. " << snd_strerror(pcm) << "\n"; logE << "ERROR: Can't set interleaved mode. " << snd_strerror(pcm) << "\n";
if ((pcm = snd_pcm_hw_params_set_format(pcm_handle, params, SND_PCM_FORMAT_S16_LE)) < 0) if ((pcm = snd_pcm_hw_params_set_format(pcm_handle_, params, SND_PCM_FORMAT_S16_LE)) < 0)
logE << "ERROR: Can't set format. " << snd_strerror(pcm) << "\n"; logE << "ERROR: Can't set format. " << snd_strerror(pcm) << "\n";
if ((pcm = snd_pcm_hw_params_set_channels(pcm_handle, params, channels)) < 0) if ((pcm = snd_pcm_hw_params_set_channels(pcm_handle_, params, channels)) < 0)
logE << "ERROR: Can't set channels number. " << snd_strerror(pcm) << "\n"; logE << "ERROR: Can't set channels number. " << snd_strerror(pcm) << "\n";
if ((pcm = snd_pcm_hw_params_set_rate_near(pcm_handle, params, &rate, 0)) < 0) if ((pcm = snd_pcm_hw_params_set_rate_near(pcm_handle_, params, &rate, 0)) < 0)
logE << "ERROR: Can't set rate. " << snd_strerror(pcm) << "\n"; logE << "ERROR: Can't set rate. " << snd_strerror(pcm) << "\n";
unsigned int buffer_time; unsigned int buffer_time;
@ -56,20 +57,20 @@ void Player::start()
unsigned int period_time = buffer_time / 4; unsigned int period_time = buffer_time / 4;
snd_pcm_hw_params_set_period_time_near(pcm_handle, params, &period_time, 0); snd_pcm_hw_params_set_period_time_near(pcm_handle_, params, &period_time, 0);
snd_pcm_hw_params_set_buffer_time_near(pcm_handle, params, &buffer_time, 0); snd_pcm_hw_params_set_buffer_time_near(pcm_handle_, params, &buffer_time, 0);
// long unsigned int periodsize = stream_->format.msRate() * 50;//2*rate/50; // long unsigned int periodsize = stream_->format.msRate() * 50;//2*rate/50;
// if ((pcm = snd_pcm_hw_params_set_buffer_size_near(pcm_handle, params, &periodsize)) < 0) // if ((pcm = snd_pcm_hw_params_set_buffer_size_near(pcm_handle, params, &periodsize)) < 0)
// logE << "Unable to set buffer size " << (long int)periodsize << ": " << snd_strerror(pcm) << "\n"; // logE << "Unable to set buffer size " << (long int)periodsize << ": " << snd_strerror(pcm) << "\n";
/* Write parameters */ /* Write parameters */
if ((pcm = snd_pcm_hw_params(pcm_handle, params)) < 0) if ((pcm = snd_pcm_hw_params(pcm_handle_, params)) < 0)
logE << "ERROR: Can't set harware parameters. " << snd_strerror(pcm) << "\n"; logE << "ERROR: Can't set harware parameters. " << snd_strerror(pcm) << "\n";
/* Resume information */ /* Resume information */
logD << "PCM name: " << snd_pcm_name(pcm_handle) << "\n"; logD << "PCM name: " << snd_pcm_name(pcm_handle_) << "\n";
logD << "PCM state: " << snd_pcm_state_name(snd_pcm_state(pcm_handle)) << "\n"; logD << "PCM state: " << snd_pcm_state_name(snd_pcm_state(pcm_handle_)) << "\n";
snd_pcm_hw_params_get_channels(params, &tmp); snd_pcm_hw_params_get_channels(params, &tmp);
logD << "channels: " << tmp << "\n"; logD << "channels: " << tmp << "\n";
@ -77,26 +78,26 @@ void Player::start()
logD << "rate: " << tmp << " bps\n"; logD << "rate: " << tmp << " bps\n";
/* Allocate buffer to hold single period */ /* Allocate buffer to hold single period */
snd_pcm_hw_params_get_period_size(params, &frames, 0); snd_pcm_hw_params_get_period_size(params, &frames_, 0);
logD << "frames: " << frames << "\n"; logD << "frames: " << frames_ << "\n";
buff_size = frames * channels * 2 /* 2 -> sample size */; buff_size = frames_ * channels * 2 /* 2 -> sample size */;
buff = (char *) malloc(buff_size); buff_ = (char *) malloc(buff_size);
snd_pcm_hw_params_get_period_time(params, &tmp, NULL); snd_pcm_hw_params_get_period_time(params, &tmp, NULL);
logD << "period time: " << tmp << "\n"; logD << "period time: " << tmp << "\n";
snd_pcm_sw_params_t *swparams; snd_pcm_sw_params_t *swparams;
snd_pcm_sw_params_alloca(&swparams); snd_pcm_sw_params_alloca(&swparams);
snd_pcm_sw_params_current(pcm_handle, swparams); snd_pcm_sw_params_current(pcm_handle_, swparams);
snd_pcm_sw_params_set_avail_min(pcm_handle, swparams, frames); snd_pcm_sw_params_set_avail_min(pcm_handle_, swparams, frames_);
snd_pcm_sw_params_set_start_threshold(pcm_handle, swparams, frames); snd_pcm_sw_params_set_start_threshold(pcm_handle_, swparams, frames_);
// snd_pcm_sw_params_set_stop_threshold(pcm_handle, swparams, frames); // snd_pcm_sw_params_set_stop_threshold(pcm_handle, swparams, frames_);
snd_pcm_sw_params(pcm_handle, swparams); snd_pcm_sw_params(pcm_handle_, swparams);
active_ = true; active_ = true;
playerThread = new thread(&Player::worker, this); playerThread_ = new thread(&Player::worker, this);
} }
@ -108,24 +109,24 @@ Player::~Player()
void Player::stop() { void Player::stop() {
active_ = false; active_ = false;
if (playerThread != NULL) if (playerThread_ != NULL)
{ {
playerThread->join(); playerThread_->join();
delete playerThread; delete playerThread_;
playerThread = NULL; playerThread_ = NULL;
} }
if (pcm_handle != NULL) if (pcm_handle_ != NULL)
{ {
snd_pcm_drain(pcm_handle); snd_pcm_drain(pcm_handle_);
snd_pcm_close(pcm_handle); snd_pcm_close(pcm_handle_);
pcm_handle = NULL; pcm_handle_ = NULL;
} }
if (buff != NULL) if (buff_ != NULL)
{ {
free(buff); free(buff_);
buff = NULL; buff_ = NULL;
} }
} }
@ -137,16 +138,16 @@ void Player::worker()
snd_pcm_sframes_t framesDelay; snd_pcm_sframes_t framesDelay;
while (active_) while (active_)
{ {
snd_pcm_avail_delay(pcm_handle, &framesAvail, &framesDelay); snd_pcm_avail_delay(pcm_handle_, &framesAvail, &framesDelay);
chronos::usec delay((chronos::usec::rep) (1000 * (double) framesDelay / stream_->format.msRate())); chronos::usec delay((chronos::usec::rep) (1000 * (double) framesDelay / stream_->getFormat().msRate()));
logD << "Avail: " << framesAvail << ", delay: " << framesDelay << ", delay[ms]: " << delay.count() / 1000 << "\n"; logD << "Avail: " << framesAvail << ", delay: " << framesDelay << ", delay[ms]: " << delay.count() / 1000 << "\n";
if (stream_->getPlayerChunk(buff, delay, frames)) if (stream_->getPlayerChunk(buff_, delay, frames_))
{ {
if ((pcm = snd_pcm_writei(pcm_handle, buff, frames)) == -EPIPE) if ((pcm = snd_pcm_writei(pcm_handle_, buff_, frames_)) == -EPIPE)
{ {
logE << "XRUN\n"; logE << "XRUN\n";
snd_pcm_prepare(pcm_handle); snd_pcm_prepare(pcm_handle_);
} }
else if (pcm < 0) else if (pcm < 0)
{ {

8
client/alsaPlayer.h
View file

@ -21,12 +21,12 @@ public:
private: private:
void worker(); void worker();
snd_pcm_t* pcm_handle; snd_pcm_t* pcm_handle_;
snd_pcm_uframes_t frames; snd_pcm_uframes_t frames_;
char *buff; char *buff_;
std::atomic<bool> active_; std::atomic<bool> active_;
Stream* stream_; Stream* stream_;
std::thread* playerThread; std::thread* playerThread_;
PcmDevice pcmDevice_; PcmDevice pcmDevice_;
}; };

64
client/clientConnection.cpp
View file

@ -10,7 +10,7 @@
using namespace std; using namespace std;
ClientConnection::ClientConnection(MessageReceiver* _receiver, const std::string& _ip, size_t _port) : active_(false), connected_(false), messageReceiver(_receiver), reqId(1), ip(_ip), port(_port), readerThread(NULL), sumTimeout(chronos::msec(0)) ClientConnection::ClientConnection(MessageReceiver* receiver, const std::string& ip, size_t port) : active_(false), connected_(false), messageReceiver_(receiver), reqId_(1), ip_(ip), port_(port), readerThread_(NULL), sumTimeout_(chronos::msec(0))
{ {
} }
@ -31,7 +31,7 @@ void ClientConnection::socketRead(void* _to, size_t _bytes)
// cout << "/"; // cout << "/";
// cout.flush(); // cout.flush();
// boost::system::error_code error; // boost::system::error_code error;
len += socket->read_some(boost::asio::buffer((char*)_to + len, toRead)); len += socket_->read_some(boost::asio::buffer((char*)_to + len, toRead));
//cout << "len: " << len << ", error: " << error << endl; //cout << "len: " << len << ", error: " << error << endl;
toRead = _bytes - len; toRead = _bytes - len;
// cout << "\\"; // cout << "\\";
@ -43,23 +43,24 @@ void ClientConnection::socketRead(void* _to, size_t _bytes)
void ClientConnection::start() void ClientConnection::start()
{ {
boost::asio::io_service io_service;
tcp::resolver resolver(io_service); tcp::resolver resolver(io_service);
tcp::resolver::query query(tcp::v4(), ip, boost::lexical_cast<string>(port)); tcp::resolver::query query(tcp::v4(), ip_, boost::lexical_cast<string>(port_));
iterator = resolver.resolve(query); auto iterator = resolver.resolve(query);
logO << "connecting\n"; logO << "connecting\n";
socket.reset(new tcp::socket(io_service)); socket_.reset(new tcp::socket(io_service));
// struct timeval tv; // struct timeval tv;
// tv.tv_sec = 5; // tv.tv_sec = 5;
// tv.tv_usec = 0; // tv.tv_usec = 0;
// cout << "socket: " << socket->native() << "\n"; // cout << "socket: " << socket->native() << "\n";
// setsockopt(socket->native(), SOL_SOCKET, SO_RCVTIMEO, &tv, sizeof(tv)); // setsockopt(socket->native(), SOL_SOCKET, SO_RCVTIMEO, &tv, sizeof(tv));
// setsockopt(socket->native(), SOL_SOCKET, SO_SNDTIMEO, &tv, sizeof(tv)); // setsockopt(socket->native(), SOL_SOCKET, SO_SNDTIMEO, &tv, sizeof(tv));
socket->connect(*iterator); socket_->connect(*iterator);
logO << "MAC: \"" << getMacAddress(socket->native()) << "\"\n"; logO << "MAC: \"" << getMacAddress(socket_->native()) << "\"\n";
connected_ = true; connected_ = true;
logS(kLogNotice) << "connected" << endl; logS(kLogNotice) << "connected" << endl;
active_ = true; active_ = true;
readerThread = new thread(&ClientConnection::reader, this); readerThread_ = new thread(&ClientConnection::reader, this);
} }
@ -70,24 +71,24 @@ void ClientConnection::stop()
try try
{ {
boost::system::error_code ec; boost::system::error_code ec;
if (socket) if (socket_)
{ {
socket->shutdown(boost::asio::ip::tcp::socket::shutdown_both, ec); socket_->shutdown(boost::asio::ip::tcp::socket::shutdown_both, ec);
if (ec) logE << "Error in socket shutdown: " << ec << "\n"; if (ec) logE << "Error in socket shutdown: " << ec << "\n";
socket->close(ec); socket_->close(ec);
if (ec) logE << "Error in socket close: " << ec << "\n"; if (ec) logE << "Error in socket close: " << ec << "\n";
} }
if (readerThread) if (readerThread_)
{ {
logD << "joining readerThread\n"; logD << "joining readerThread\n";
readerThread->join(); readerThread_->join();
delete readerThread; delete readerThread_;
} }
} }
catch(...) catch(...)
{ {
} }
readerThread = NULL; readerThread_ = NULL;
logD << "readerThread terminated\n"; logD << "readerThread terminated\n";
} }
@ -104,7 +105,7 @@ bool ClientConnection::send(msg::BaseMessage* message)
tv t; tv t;
message->sent = t; message->sent = t;
message->serialize(stream); message->serialize(stream);
boost::asio::write(*socket.get(), streambuf); boost::asio::write(*socket_.get(), streambuf);
return true; return true;
} }
@ -112,34 +113,35 @@ bool ClientConnection::send(msg::BaseMessage* message)
shared_ptr<msg::SerializedMessage> ClientConnection::sendRequest(msg::BaseMessage* message, const chronos::msec& timeout) shared_ptr<msg::SerializedMessage> ClientConnection::sendRequest(msg::BaseMessage* message, const chronos::msec& timeout)
{ {
shared_ptr<msg::SerializedMessage> response(NULL); shared_ptr<msg::SerializedMessage> response(NULL);
if (++reqId == 10000) if (++reqId_ == 10000)
reqId = 1; reqId_ = 1;
message->id = reqId; message->id = reqId_;
//logD << "Req: " << reqId << "\n"; //logD << "Req: " << reqId << "\n";
shared_ptr<PendingRequest> pendingRequest(new PendingRequest(reqId)); shared_ptr<PendingRequest> pendingRequest(new PendingRequest(reqId_));
{ {
std::unique_lock<std::mutex> mlock(mutex_); std::unique_lock<std::mutex> mlock(mutex_);
pendingRequests.insert(pendingRequest); pendingRequests_.insert(pendingRequest);
} }
std::mutex m;
std::unique_lock<std::mutex> lck(m); std::unique_lock<std::mutex> lck(m);
send(message); send(message);
if (pendingRequest->cv.wait_for(lck,std::chrono::milliseconds(timeout)) == std::cv_status::no_timeout) if (pendingRequest->cv.wait_for(lck,std::chrono::milliseconds(timeout)) == std::cv_status::no_timeout)
{ {
response = pendingRequest->response; response = pendingRequest->response;
sumTimeout = chronos::msec(0); sumTimeout_ = chronos::msec(0);
//logD << "Resp: " << pendingRequest->id << "\n"; //logD << "Resp: " << pendingRequest->id << "\n";
} }
else else
{ {
sumTimeout += timeout; sumTimeout_ += timeout;
logO << "timeout while waiting for response to: " << reqId << ", timeout " << sumTimeout.count() << "\n"; logO << "timeout while waiting for response to: " << reqId_ << ", timeout " << sumTimeout_.count() << "\n";
if (sumTimeout > chronos::sec(10)) if (sumTimeout_ > chronos::sec(10))
throw SnapException("sum timeout exceeded 10s"); throw SnapException("sum timeout exceeded 10s");
} }
{ {
std::unique_lock<std::mutex> mlock(mutex_); std::unique_lock<std::mutex> mlock(mutex_);
pendingRequests.erase(pendingRequest); pendingRequests_.erase(pendingRequest);
} }
return response; return response;
} }
@ -162,7 +164,7 @@ void ClientConnection::getNextMessage()
{ {
std::unique_lock<std::mutex> mlock(mutex_); std::unique_lock<std::mutex> mlock(mutex_);
{ {
for (auto req: pendingRequests) for (auto req: pendingRequests_)
{ {
if (req->id == baseMessage.refersTo) if (req->id == baseMessage.refersTo)
{ {
@ -177,8 +179,8 @@ void ClientConnection::getNextMessage()
} }
} }
if (messageReceiver != NULL) if (messageReceiver_ != NULL)
messageReceiver->onMessageReceived(this, baseMessage, &buffer[0]); messageReceiver_->onMessageReceived(this, baseMessage, &buffer[0]);
} }
@ -194,8 +196,8 @@ void ClientConnection::reader()
} }
catch (const std::exception& e) catch (const std::exception& e)
{ {
if (messageReceiver != NULL) if (messageReceiver_ != NULL)
messageReceiver->onException(this, e); messageReceiver_->onException(this, e);
} }
catch (...) catch (...)
{ {

30
client/clientConnection.h
View file

@ -40,11 +40,11 @@ public:
class ClientConnection class ClientConnection
{ {
public: public:
ClientConnection(MessageReceiver* _receiver, const std::string& _ip, size_t _port); ClientConnection(MessageReceiver* receiver, const std::string& ip, size_t port);
virtual ~ClientConnection(); virtual ~ClientConnection();
virtual void start(); virtual void start();
virtual void stop(); virtual void stop();
virtual bool send(msg::BaseMessage* _message); virtual bool send(msg::BaseMessage* message);
virtual std::shared_ptr<msg::SerializedMessage> sendRequest(msg::BaseMessage* message, const chronos::msec& timeout = chronos::msec(1000)); virtual std::shared_ptr<msg::SerializedMessage> sendRequest(msg::BaseMessage* message, const chronos::msec& timeout = chronos::msec(1000));
template <typename T> template <typename T>
@ -65,31 +65,27 @@ public:
virtual bool connected() virtual bool connected()
{ {
return (socket != 0); return (socket_ != 0);
// return (connected_ && socket); // return (connected_ && socket);
} }
protected: protected:
virtual void reader(); virtual void reader();
void socketRead(void* _to, size_t _bytes); void socketRead(void* to, size_t bytes);
std::shared_ptr<tcp::socket> socket; void getNextMessage();
// boost::asio::ip::tcp::endpoint endpt; std::shared_ptr<tcp::socket> socket_;
std::atomic<bool> active_; std::atomic<bool> active_;
std::atomic<bool> connected_; std::atomic<bool> connected_;
MessageReceiver* messageReceiver; MessageReceiver* messageReceiver_;
void getNextMessage();
boost::asio::io_service io_service;
tcp::resolver::iterator iterator;
mutable std::mutex mutex_; mutable std::mutex mutex_;
std::mutex m; std::set<std::shared_ptr<PendingRequest>> pendingRequests_;
std::set<std::shared_ptr<PendingRequest>> pendingRequests; uint16_t reqId_;
uint16_t reqId; std::string ip_;
std::string ip; size_t port_;
size_t port; std::thread* readerThread_;
std::thread* readerThread; chronos::msec sumTimeout_;
chronos::msec sumTimeout;
}; };

73
client/controller.cpp
View file

@ -18,7 +18,7 @@
using namespace std; using namespace std;
Controller::Controller() : MessageReceiver(), active_(false), sampleFormat(NULL), decoder(NULL) Controller::Controller() : MessageReceiver(), active_(false), sampleFormat_(NULL), decoder_(NULL)
{ {
} }
@ -33,14 +33,14 @@ void Controller::onMessageReceived(ClientConnection* connection, const msg::Base
{ {
if (baseMessage.type == message_type::kPayload) if (baseMessage.type == message_type::kPayload)
{ {
if ((stream != NULL) && (decoder != NULL)) if ((stream_ != NULL) && (decoder_ != NULL))
{ {
msg::PcmChunk* pcmChunk = new msg::PcmChunk(*sampleFormat, 0); msg::PcmChunk* pcmChunk = new msg::PcmChunk(*sampleFormat_, 0);
pcmChunk->deserialize(baseMessage, buffer); pcmChunk->deserialize(baseMessage, buffer);
//logD << "chunk: " << pcmChunk->payloadSize; //logD << "chunk: " << pcmChunk->payloadSize;
if (decoder->decode(pcmChunk)) if (decoder_->decode(pcmChunk))
{ {
stream->addChunk(pcmChunk); stream_->addChunk(pcmChunk);
//logD << ", decoded: " << pcmChunk->payloadSize << ", Duration: " << pcmChunk->getDuration() << ", sec: " << pcmChunk->timestamp.sec << ", usec: " << pcmChunk->timestamp.usec/1000 << ", type: " << pcmChunk->type << "\n"; //logD << ", decoded: " << pcmChunk->payloadSize << ", Duration: " << pcmChunk->getDuration() << ", sec: " << pcmChunk->timestamp.sec << ", usec: " << pcmChunk->timestamp.usec/1000 << ", type: " << pcmChunk->type << "\n";
} }
else else
@ -50,13 +50,13 @@ void Controller::onMessageReceived(ClientConnection* connection, const msg::Base
} }
void Controller::start(const PcmDevice& pcmDevice, const std::string& _ip, size_t _port, size_t latency) void Controller::start(const PcmDevice& pcmDevice, const std::string& ip, size_t port, size_t latency)
{ {
ip = _ip; ip_ = ip;
pcmDevice_ = pcmDevice; pcmDevice_ = pcmDevice;
latency_ = latency; latency_ = latency;
clientConnection = new ClientConnection(this, ip, _port); clientConnection_ = new ClientConnection(this, ip, port);
controllerThread = new thread(&Controller::worker, this); controllerThread_ = new thread(&Controller::worker, this);
} }
@ -64,48 +64,47 @@ void Controller::stop()
{ {
logD << "Stopping\n"; logD << "Stopping\n";
active_ = false; active_ = false;
controllerThread->join(); controllerThread_->join();
clientConnection->stop(); clientConnection_->stop();
delete controllerThread; delete controllerThread_;
delete clientConnection; delete clientConnection_;
} }
void Controller::worker() void Controller::worker()
{ {
// Decoder* decoder;
active_ = true; active_ = true;
decoder = NULL; decoder_ = NULL;
stream = NULL; stream_ = NULL;
while (active_) while (active_)
{ {
try try
{ {
clientConnection->start(); clientConnection_->start();
msg::Request requestMsg(kServerSettings); msg::Request requestMsg(kServerSettings);
shared_ptr<msg::ServerSettings> serverSettings(NULL); shared_ptr<msg::ServerSettings> serverSettings(NULL);
while (active_ && !(serverSettings = clientConnection->sendReq<msg::ServerSettings>(&requestMsg))); while (active_ && !(serverSettings = clientConnection_->sendReq<msg::ServerSettings>(&requestMsg)));
logO << "ServerSettings buffer: " << serverSettings->bufferMs << "\n"; logO << "ServerSettings buffer: " << serverSettings->bufferMs << "\n";
requestMsg.request = kSampleFormat; requestMsg.request = kSampleFormat;
while (active_ && !(sampleFormat = clientConnection->sendReq<msg::SampleFormat>(&requestMsg))); while (active_ && !(sampleFormat_ = clientConnection_->sendReq<msg::SampleFormat>(&requestMsg)));
logO << "SampleFormat rate: " << sampleFormat->rate << ", bits: " << sampleFormat->bits << ", channels: " << sampleFormat->channels << "\n"; logO << "SampleFormat rate: " << sampleFormat_->rate << ", bits: " << sampleFormat_->bits << ", channels: " << sampleFormat_->channels << "\n";
requestMsg.request = kHeader; requestMsg.request = kHeader;
shared_ptr<msg::Header> headerChunk(NULL); shared_ptr<msg::Header> headerChunk(NULL);
while (active_ && !(headerChunk = clientConnection->sendReq<msg::Header>(&requestMsg))); while (active_ && !(headerChunk = clientConnection_->sendReq<msg::Header>(&requestMsg)));
logO << "Codec: " << headerChunk->codec << "\n"; logO << "Codec: " << headerChunk->codec << "\n";
if (headerChunk->codec == "ogg") if (headerChunk->codec == "ogg")
decoder = new OggDecoder(); decoder_ = new OggDecoder();
else if (headerChunk->codec == "pcm") else if (headerChunk->codec == "pcm")
decoder = new PcmDecoder(); decoder_ = new PcmDecoder();
decoder->setHeader(headerChunk.get()); decoder_->setHeader(headerChunk.get());
msg::Request timeReq(kTime); msg::Request timeReq(kTime);
for (size_t n=0; n<50 && active_; ++n) for (size_t n=0; n<50 && active_; ++n)
{ {
shared_ptr<msg::Time> reply = clientConnection->sendReq<msg::Time>(&timeReq, chronos::msec(2000)); shared_ptr<msg::Time> reply = clientConnection_->sendReq<msg::Time>(&timeReq, chronos::msec(2000));
if (reply) if (reply)
{ {
double latency = (reply->received.sec - reply->sent.sec) + (reply->received.usec - reply->sent.usec) / 1000000.; double latency = (reply->received.sec - reply->sent.sec) + (reply->received.usec - reply->sent.usec) / 1000000.;
@ -115,21 +114,21 @@ void Controller::worker()
} }
logO << "diff to server [ms]: " << TimeProvider::getInstance().getDiffToServer<chronos::msec>().count() << "\n"; logO << "diff to server [ms]: " << TimeProvider::getInstance().getDiffToServer<chronos::msec>().count() << "\n";
stream = new Stream(*sampleFormat); stream_ = new Stream(*sampleFormat_);
stream->setBufferLen(serverSettings->bufferMs - latency_); stream_->setBufferLen(serverSettings->bufferMs - latency_);
Player player(pcmDevice_, stream); Player player(pcmDevice_, stream_);
player.start(); player.start();
msg::Command startStream("startStream"); msg::Command startStream("startStream");
shared_ptr<msg::Ack> ackMsg(NULL); shared_ptr<msg::Ack> ackMsg(NULL);
while (active_ && !(ackMsg = clientConnection->sendReq<msg::Ack>(&startStream))); while (active_ && !(ackMsg = clientConnection_->sendReq<msg::Ack>(&startStream)));
while (active_) while (active_)
{ {
usleep(500*1000); usleep(500*1000);
//throw SnapException("timeout"); //throw SnapException("timeout");
shared_ptr<msg::Time> reply = clientConnection->sendReq<msg::Time>(&timeReq); shared_ptr<msg::Time> reply = clientConnection_->sendReq<msg::Time>(&timeReq);
if (reply) if (reply)
{ {
double latency = (reply->received.sec - reply->sent.sec) + (reply->received.usec - reply->sent.usec) / 1000000.; double latency = (reply->received.sec - reply->sent.sec) + (reply->received.usec - reply->sent.usec) / 1000000.;
@ -141,14 +140,14 @@ void Controller::worker()
{ {
logS(kLogErr) << "Exception in Controller::worker(): " << e.what() << endl; logS(kLogErr) << "Exception in Controller::worker(): " << e.what() << endl;
logO << "Stopping clientConnection" << endl; logO << "Stopping clientConnection" << endl;
clientConnection->stop(); clientConnection_->stop();
logO << "Deleting stream" << endl; logO << "Deleting stream" << endl;
if (stream != NULL) if (stream_ != NULL)
delete stream; delete stream_;
stream = NULL; stream_ = NULL;
if (decoder != NULL) if (decoder_ != NULL)
delete decoder; delete decoder_;
decoder = NULL; decoder_ = NULL;
logO << "done" << endl; logO << "done" << endl;
if (active_) if (active_)
usleep(500*1000); usleep(500*1000);

14
client/controller.h
View file

@ -14,7 +14,7 @@ class Controller : public MessageReceiver
{ {
public: public:
Controller(); Controller();
void start(const PcmDevice& pcmDevice, const std::string& _ip, size_t _port, size_t latency); void start(const PcmDevice& pcmDevice, const std::string& ip, size_t port, size_t latency);
void stop(); void stop();
virtual void onMessageReceived(ClientConnection* connection, const msg::BaseMessage& baseMessage, char* buffer); virtual void onMessageReceived(ClientConnection* connection, const msg::BaseMessage& baseMessage, char* buffer);
virtual void onException(ClientConnection* connection, const std::exception& exception); virtual void onException(ClientConnection* connection, const std::exception& exception);
@ -22,12 +22,12 @@ public:
private: private:
void worker(); void worker();
std::atomic<bool> active_; std::atomic<bool> active_;
std::thread* controllerThread; std::thread* controllerThread_;
ClientConnection* clientConnection; ClientConnection* clientConnection_;
Stream* stream; Stream* stream_;
std::string ip; std::string ip_;
std::shared_ptr<msg::SampleFormat> sampleFormat; std::shared_ptr<msg::SampleFormat> sampleFormat_;
Decoder* decoder; Decoder* decoder_;
PcmDevice pcmDevice_; PcmDevice pcmDevice_;
size_t latency_; size_t latency_;
}; };

224
client/stream.cpp
View file

@ -8,13 +8,13 @@ using namespace std;
using namespace chronos; using namespace chronos;
Stream::Stream(const msg::SampleFormat& sampleFormat) : format(format_), format_(sampleFormat), sleep(0), median(0), shortMedian(0), lastUpdate(0), playedFrames(0) Stream::Stream(const msg::SampleFormat& sampleFormat) : format_(sampleFormat), sleep_(0), median_(0), shortMedian_(0), lastUpdate_(0), playedFrames_(0)
{ {
buffer.setSize(500); buffer_.setSize(500);
shortBuffer.setSize(100); shortBuffer_.setSize(100);
miniBuffer.setSize(20); miniBuffer_.setSize(20);
// cardBuffer.setSize(50); // cardBuffer_.setSize(50);
bufferMs = msec(500); bufferMs_ = msec(500);
/* /*
48000 x 48000 x
@ -23,50 +23,50 @@ Stream::Stream(const msg::SampleFormat& sampleFormat) : format(format_), format_
x = 1,000016667 / (1,000016667 - 1) x = 1,000016667 / (1,000016667 - 1)
*/ */
setRealSampleRate(format.rate); setRealSampleRate(format_.rate);
} }
void Stream::setRealSampleRate(double sampleRate) void Stream::setRealSampleRate(double sampleRate)
{ {
if (sampleRate == format.rate) if (sampleRate == format_.rate)
correctAfterXFrames = 0; correctAfterXFrames_ = 0;
else else
correctAfterXFrames = round((format.rate / sampleRate) / (format.rate / sampleRate - 1.)); correctAfterXFrames_ = round((format_.rate / sampleRate) / (format_.rate / sampleRate - 1.));
// logD << "Correct after X: " << correctAfterXFrames << " (Real rate: " << sampleRate << ", rate: " << format.rate << ")\n"; // logD << "Correct after X: " << correctAfterXFrames_ << " (Real rate: " << sampleRate << ", rate: " << format_.rate << ")\n";
} }
void Stream::setBufferLen(size_t bufferLenMs) void Stream::setBufferLen(size_t bufferLenMs)
{ {
bufferMs = msec(bufferLenMs); bufferMs_ = msec(bufferLenMs);
} }
void Stream::clearChunks() void Stream::clearChunks()
{ {
while (chunks.size() > 0) while (chunks_.size() > 0)
chunks.pop(); chunks_.pop();
} }
void Stream::addChunk(msg::PcmChunk* chunk) void Stream::addChunk(msg::PcmChunk* chunk)
{ {
while (chunks.size() * chunk->duration<chronos::msec>().count() > 10000) while (chunks_.size() * chunk_->duration<chronos::msec>().count() > 10000)
chunks.pop(); chunks_.pop();
chunks.push(shared_ptr<msg::PcmChunk>(chunk)); chunks_.push(shared_ptr<msg::PcmChunk>(chunk));
// logD << "new chunk: " << chunk->getDuration() << ", Chunks: " << chunks.size() << "\n"; // logD << "new chunk: " << chunk_->getDuration() << ", Chunks: " << chunks_.size() << "\n";
} }
time_point_hrc Stream::getSilentPlayerChunk(void* outputBuffer, unsigned long framesPerBuffer) time_point_hrc Stream::getSilentPlayerChunk(void* outputBuffer, unsigned long framesPerBuffer)
{ {
if (!chunk) if (!chunk_)
chunk = chunks.pop(); chunk_ = chunks_.pop();
time_point_hrc tp = chunk->start(); time_point_hrc tp = chunk_->start();
memset(outputBuffer, 0, framesPerBuffer * format.frameSize); memset(outputBuffer, 0, framesPerBuffer * format_.frameSize);
return tp; return tp;
} }
@ -75,16 +75,16 @@ time_point_hrc Stream::getSilentPlayerChunk(void* outputBuffer, unsigned long fr
time_point_ms Stream::seekTo(const time_point_ms& to) time_point_ms Stream::seekTo(const time_point_ms& to)
{ {
if (!chunk) if (!chunk)
chunk = chunks.pop(); chunk_ = chunks_.pop();
// time_point_ms tp = chunk->timePoint(); // time_point_ms tp = chunk_->timePoint();
while (to > chunk->timePoint())// + std::chrono::milliseconds((long int)chunk->getDuration()))// while (to > chunk_->timePoint())// + std::chrono::milliseconds((long int)chunk_->getDuration()))//
{ {
chunk = chunks.pop(); chunk_ = chunks_.pop();
logD << "\nto: " << Chunk::getAge(to) << "\t chunk: " << Chunk::getAge(chunk->timePoint()) << "\n"; logD << "\nto: " << Chunk::getAge(to) << "\t chunk: " << Chunk::getAge(chunk_->timePoint()) << "\n";
logD << "diff: " << std::chrono::duration_cast<std::chrono::milliseconds>((to - chunk->timePoint())).count() << "\n"; logD << "diff: " << std::chrono::duration_cast<std::chrono::milliseconds>((to - chunk_->timePoint())).count() << "\n";
} }
chunk->seek(std::chrono::duration_cast<std::chrono::milliseconds>(to - chunk->timePoint()).count() * format.msRate()); chunk_->seek(std::chrono::duration_cast<std::chrono::milliseconds>(to - chunk_->timePoint()).count() * format_.msRate());
return chunk->timePoint(); return chunk_->timePoint();
} }
*/ */
@ -92,35 +92,35 @@ time_point_ms Stream::seekTo(const time_point_ms& to)
time_point_hrc Stream::seek(long ms) time_point_hrc Stream::seek(long ms)
{ {
if (!chunk) if (!chunk)
chunk = chunks.pop(); chunk_ = chunks_.pop();
if (ms <= 0) if (ms <= 0)
return chunk->start(); return chunk_->start();
// time_point_ms tp = chunk->timePoint(); // time_point_ms tp = chunk_->timePoint();
while (ms > chunk->duration<chronos::msec>().count()) while (ms > chunk_->duration<chronos::msec>().count())
{ {
chunk = chunks.pop(); chunk_ = chunks_.pop();
ms -= min(ms, (long)chunk->durationLeft<chronos::msec>().count()); ms -= min(ms, (long)chunk_->durationLeft<chronos::msec>().count());
} }
chunk->seek(ms * format.msRate()); chunk_->seek(ms * format_.msRate());
return chunk->start(); return chunk_->start();
} }
*/ */
time_point_hrc Stream::getNextPlayerChunk(void* outputBuffer, const chronos::usec& timeout, unsigned long framesPerBuffer) time_point_hrc Stream::getNextPlayerChunk(void* outputBuffer, const chronos::usec& timeout, unsigned long framesPerBuffer)
{ {
if (!chunk && !chunks.try_pop(chunk, timeout)) if (!chunk_ && !chunks_.try_pop(chunk_, timeout))
throw 0; throw 0;
time_point_hrc tp = chunk->start(); time_point_hrc tp = chunk_->start();
char* buffer = (char*)outputBuffer; char* buffer = (char*)outputBuffer;
unsigned long read = 0; unsigned long read = 0;
while (read < framesPerBuffer) while (read < framesPerBuffer)
{ {
read += chunk->readFrames(buffer + read*format.frameSize, framesPerBuffer - read); read += chunk_->readFrames(buffer + read*format_.frameSize, framesPerBuffer - read);
if (chunk->isEndOfChunk() && !chunks.try_pop(chunk, timeout)) if (chunk_->isEndOfChunk() && !chunks_.try_pop(chunk_, timeout))
throw 0; throw 0;
} }
return tp; return tp;
@ -133,7 +133,7 @@ time_point_hrc Stream::getNextPlayerChunk(void* outputBuffer, const chronos::use
return getNextPlayerChunk(outputBuffer, timeout, framesPerBuffer); return getNextPlayerChunk(outputBuffer, timeout, framesPerBuffer);
long toRead = framesPerBuffer + framesCorrection; long toRead = framesPerBuffer + framesCorrection;
char* buffer = (char*)malloc(toRead * format.frameSize); char* buffer = (char*)malloc(toRead * format_.frameSize);
time_point_hrc tp = getNextPlayerChunk(buffer, timeout, toRead); time_point_hrc tp = getNextPlayerChunk(buffer, timeout, toRead);
float factor = (float)toRead / framesPerBuffer;//(float)(framesPerBuffer*channels_); float factor = (float)toRead / framesPerBuffer;//(float)(framesPerBuffer*channels_);
@ -143,7 +143,7 @@ time_point_hrc Stream::getNextPlayerChunk(void* outputBuffer, const chronos::use
for (size_t n=0; n<framesPerBuffer; ++n) for (size_t n=0; n<framesPerBuffer; ++n)
{ {
size_t index(floor(idx));// = (int)(ceil(n*factor)); size_t index(floor(idx));// = (int)(ceil(n*factor));
memcpy((char*)outputBuffer + n*format.frameSize, buffer + index*format.frameSize, format.frameSize); memcpy((char*)outputBuffer + n*format_.frameSize, buffer + index*format_.frameSize, format_.frameSize);
idx += factor; idx += factor;
} }
free(buffer); free(buffer);
@ -155,36 +155,36 @@ time_point_hrc Stream::getNextPlayerChunk(void* outputBuffer, const chronos::use
void Stream::updateBuffers(int age) void Stream::updateBuffers(int age)
{ {
buffer.add(age); buffer_.add(age);
miniBuffer.add(age); miniBuffer_.add(age);
shortBuffer.add(age); shortBuffer_.add(age);
} }
void Stream::resetBuffers() void Stream::resetBuffers()
{ {
buffer.clear(); buffer_.clear();
miniBuffer.clear(); miniBuffer_.clear();
shortBuffer.clear(); shortBuffer_.clear();
} }
bool Stream::getPlayerChunk(void* outputBuffer, const chronos::usec& outputBufferDacTime, unsigned long framesPerBuffer) bool Stream::getPlayerChunk(void* outputBuffer, const chronos::usec& outputBufferDacTime, unsigned long framesPerBuffer)
{ {
if (outputBufferDacTime > bufferMs) if (outputBufferDacTime > bufferMs_)
return false; return false;
if (!chunk && !chunks.try_pop(chunk, outputBufferDacTime)) if (!chunk_ && !chunks_.try_pop(chunk_, outputBufferDacTime))
return false; return false;
playedFrames += framesPerBuffer; playedFrames_ += framesPerBuffer;
chronos::usec age = std::chrono::duration_cast<usec>(TimeProvider::serverNow() - chunk->start() - bufferMs + outputBufferDacTime); chronos::usec age = std::chrono::duration_cast<usec>(TimeProvider::serverNow() - chunk_->start() - bufferMs_ + outputBufferDacTime);
if ((sleep.count() == 0) && (chronos::abs(age) > chronos::msec(200))) if ((sleep_.count() == 0) && (chronos::abs(age) > chronos::msec(200)))
{ {
logO << "age > 200: " << age.count() << "\n"; logO << "age > 200: " << age.count() << "\n";
sleep = age; sleep_ = age;
} }
try try
@ -195,126 +195,126 @@ bool Stream::getPlayerChunk(void* outputBuffer, const chronos::usec& outputBuffe
// logD << "buffer duration: " << bufferDuration.count() << "\n"; // logD << "buffer duration: " << bufferDuration.count() << "\n";
chronos::usec correction = chronos::usec(0); chronos::usec correction = chronos::usec(0);
if (sleep.count() != 0) if (sleep_.count() != 0)
{ {
resetBuffers(); resetBuffers();
if (sleep < -bufferDuration/2) if (sleep_ < -bufferDuration/2)
{ {
// We're early: not enough chunks. play silence. Reference chunk is the oldest (front) one // We're early: not enough chunks_. play silence. Reference chunk_ is the oldest (front) one
sleep = chrono::duration_cast<usec>(TimeProvider::serverNow() - getSilentPlayerChunk(outputBuffer, framesPerBuffer) - bufferMs + outputBufferDacTime); sleep_ = chrono::duration_cast<usec>(TimeProvider::serverNow() - getSilentPlayerChunk(outputBuffer, framesPerBuffer) - bufferMs_ + outputBufferDacTime);
//logD << "-sleep: " << sleep.count() << " " << -bufferDuration.count() / 2000 << "\n"; //logD << "-sleep: " << sleep_.count() << " " << -bufferDuration.count() / 2000 << "\n";
if (sleep < -bufferDuration/2) if (sleep_ < -bufferDuration/2)
return true; return true;
} }
else if (sleep > bufferDuration/2) else if (sleep_ > bufferDuration/2)
{ {
// We're late: discard oldest chunks // We're late: discard oldest chunks
while (sleep > chunk->duration<chronos::usec>()) while (sleep_ > chunk_->duration<chronos::usec>())
{ {
logO << "sleep > chunk->getDuration(): " << sleep.count() << " > " << chunk->duration<chronos::msec>().count() << ", chunks: " << chunks.size() << ", out: " << outputBufferDacTime.count() << ", needed: " << bufferDuration.count() << "\n"; logO << "sleep > chunk_->getDuration(): " << sleep_.count() << " > " << chunk_->duration<chronos::msec>().count() << ", chunks: " << chunks_.size() << ", out: " << outputBufferDacTime.count() << ", needed: " << bufferDuration.count() << "\n";
sleep = std::chrono::duration_cast<usec>(TimeProvider::serverNow() - chunk->start() - bufferMs + outputBufferDacTime); sleep_ = std::chrono::duration_cast<usec>(TimeProvider::serverNow() - chunk_->start() - bufferMs_ + outputBufferDacTime);
if (!chunks.try_pop(chunk, outputBufferDacTime)) if (!chunks_.try_pop(chunk_, outputBufferDacTime))
{ {
logO << "no chunks available\n"; logO << "no chunks available\n";
chunk = NULL; chunk_ = NULL;
sleep = chronos::usec(0); sleep_ = chronos::usec(0);
return false; return false;
} }
} }
} }
// out of sync, can be corrected by playing faster/slower // out of sync, can be corrected by playing faster/slower
if (sleep < -chronos::usec(100)) if (sleep_ < -chronos::usec(100))
{ {
sleep += chronos::usec(100); sleep_ += chronos::usec(100);
correction = -chronos::usec(100); correction = -chronos::usec(100);
} }
else if (sleep > chronos::usec(100)) else if (sleep_ > chronos::usec(100))
{ {
sleep -= chronos::usec(100); sleep_ -= chronos::usec(100);
correction = chronos::usec(100); correction = chronos::usec(100);
} }
else else
{ {
logO << "Sleep " << sleep.count() << "\n"; logO << "Sleep " << sleep_.count() << "\n";
correction = sleep; correction = sleep_;
sleep = chronos::usec(0); sleep_ = chronos::usec(0);
} }
} }
long framesCorrection = correction.count()*format.usRate(); long framesCorrection = correction.count()*format_.usRate();
if ((correctAfterXFrames != 0) && (playedFrames >= (unsigned long)abs(correctAfterXFrames))) if ((correctAfterXFrames_ != 0) && (playedFrames_ >= (unsigned long)abs(correctAfterXFrames_)))
{ {
framesCorrection += (correctAfterXFrames > 0)?1:-1; framesCorrection += (correctAfterXFrames_ > 0)?1:-1;
playedFrames -= abs(correctAfterXFrames); playedFrames_ -= abs(correctAfterXFrames_);
} }
age = std::chrono::duration_cast<usec>(TimeProvider::serverNow() - getNextPlayerChunk(outputBuffer, outputBufferDacTime, framesPerBuffer, framesCorrection) - bufferMs + outputBufferDacTime); age = std::chrono::duration_cast<usec>(TimeProvider::serverNow() - getNextPlayerChunk(outputBuffer, outputBufferDacTime, framesPerBuffer, framesCorrection) - bufferMs_ + outputBufferDacTime);
setRealSampleRate(format.rate); setRealSampleRate(format_.rate);
if (sleep.count() == 0) if (sleep_.count() == 0)
{ {
if (buffer.full()) if (buffer_.full())
{ {
if (chronos::usec(abs(median)) > chronos::msec(1)) if (chronos::usec(abs(median_)) > chronos::msec(1))
{ {
logO << "pBuffer->full() && (abs(median) > 1): " << median << "\n"; logO << "pBuffer->full() && (abs(median_) > 1): " << median_ << "\n";
sleep = chronos::usec(shortMedian); sleep_ = chronos::usec(shortMedian_);
} }
/* else if (chronos::usec(median) > chronos::usec(300)) /* else if (chronos::usec(median_) > chronos::usec(300))
{ {
setRealSampleRate(format.rate - format.rate / 1000); setRealSampleRate(format_.rate - format_.rate / 1000);
} }
else if (chronos::usec(median) < -chronos::usec(300)) else if (chronos::usec(median_) < -chronos::usec(300))
{ {
setRealSampleRate(format.rate + format.rate / 1000); setRealSampleRate(format_.rate + format_.rate / 1000);
} }
*/ } */ }
else if (shortBuffer.full()) else if (shortBuffer_.full())
{ {
if (chronos::usec(abs(shortMedian)) > chronos::msec(5)) if (chronos::usec(abs(shortMedian_)) > chronos::msec(5))
{ {
logO << "pShortBuffer->full() && (abs(shortMedian) > 5): " << shortMedian << "\n"; logO << "pShortBuffer->full() && (abs(shortMedian_) > 5): " << shortMedian_ << "\n";
sleep = chronos::usec(shortMedian); sleep_ = chronos::usec(shortMedian_);
} }
/* else /* else
{ {
setRealSampleRate(format.rate + -shortMedian / 100); setRealSampleRate(format_.rate + -shortMedian_ / 100);
} }
*/ } */ }
else if (miniBuffer.full() && (chronos::usec(abs(miniBuffer.median())) > chronos::msec(50))) else if (miniBuffer_.full() && (chronos::usec(abs(miniBuffer_.median())) > chronos::msec(50)))
{ {
logO << "pMiniBuffer->full() && (abs(pMiniBuffer->mean()) > 50): " << miniBuffer.median() << "\n"; logO << "pMiniBuffer->full() && (abs(pMiniBuffer->mean()) > 50): " << miniBuffer_.median() << "\n";
sleep = chronos::usec((chronos::msec::rep)miniBuffer.mean()); sleep_ = chronos::usec((chronos::msec::rep)miniBuffer_.mean());
} }
} }
if (sleep.count() != 0) if (sleep_.count() != 0)
logO << "Sleep: " << sleep.count() << "\n"; logO << "Sleep: " << sleep_.count() << "\n";
else if (shortBuffer.full()) else if (shortBuffer_.full())
{ {
if (chronos::usec(shortMedian) > chronos::usec(100)) if (chronos::usec(shortMedian_) > chronos::usec(100))
setRealSampleRate(format.rate * 0.9999); setRealSampleRate(format_.rate * 0.9999);
else if (chronos::usec(shortMedian) < -chronos::usec(100)) else if (chronos::usec(shortMedian_) < -chronos::usec(100))
setRealSampleRate(format.rate * 1.0001); setRealSampleRate(format_.rate * 1.0001);
} }
updateBuffers(age.count()); updateBuffers(age.count());
// print sync stats // print sync stats
time_t now = time(NULL); time_t now = time(NULL);
if (now != lastUpdate) if (now != lastUpdate_)
{ {
lastUpdate = now; lastUpdate_ = now;
median = buffer.median(); median_ = buffer_.median();
shortMedian = shortBuffer.median(); shortMedian_ = shortBuffer_.median();
logO << "Chunk: " << age.count()/100 << "\t" << miniBuffer.median()/100 << "\t" << shortMedian/100 << "\t" << median/100 << "\t" << buffer.size() << "\t" << outputBufferDacTime.count() << "\n"; logO << "Chunk: " << age.count()/100 << "\t" << miniBuffer_.median()/100 << "\t" << shortMedian_/100 << "\t" << median_/100 << "\t" << buffer_.size() << "\t" << outputBufferDacTime.count() << "\n";
} }
return true; return true;
} }
catch(int e) catch(int e)
{ {
sleep = chronos::usec(0); sleep_ = chronos::usec(0);
return false; return false;
} }
} }

31
client/stream.h
View file

@ -24,7 +24,10 @@ public:
void clearChunks(); void clearChunks();
bool getPlayerChunk(void* outputBuffer, const chronos::usec& outputBufferDacTime, unsigned long framesPerBuffer); bool getPlayerChunk(void* outputBuffer, const chronos::usec& outputBufferDacTime, unsigned long framesPerBuffer);
void setBufferLen(size_t bufferLenMs); void setBufferLen(size_t bufferLenMs);
const msg::SampleFormat& format; const msg::SampleFormat& getFormat() const
{
return format_;
}
private: private:
chronos::time_point_hrc getNextPlayerChunk(void* outputBuffer, const chronos::usec& timeout, unsigned long framesPerBuffer); chronos::time_point_hrc getNextPlayerChunk(void* outputBuffer, const chronos::usec& timeout, unsigned long framesPerBuffer);
@ -38,22 +41,22 @@ private:
msg::SampleFormat format_; msg::SampleFormat format_;
long lastTick; long lastTick_;
chronos::usec sleep; chronos::usec sleep_;
Queue<std::shared_ptr<msg::PcmChunk>> chunks; Queue<std::shared_ptr<msg::PcmChunk>> chunks_;
// DoubleBuffer<chronos::usec::rep> cardBuffer; // DoubleBuffer<chronos::usec::rep> cardBuffer;
DoubleBuffer<chronos::usec::rep> miniBuffer; DoubleBuffer<chronos::usec::rep> miniBuffer_;
DoubleBuffer<chronos::usec::rep> buffer; DoubleBuffer<chronos::usec::rep> buffer_;
DoubleBuffer<chronos::usec::rep> shortBuffer; DoubleBuffer<chronos::usec::rep> shortBuffer_;
std::shared_ptr<msg::PcmChunk> chunk; std::shared_ptr<msg::PcmChunk> chunk_;
int median; int median_;
int shortMedian; int shortMedian_;
time_t lastUpdate; time_t lastUpdate_;
chronos::msec bufferMs; chronos::msec bufferMs_;
unsigned long playedFrames; unsigned long playedFrames_;
long correctAfterXFrames; long correctAfterXFrames_;
}; };

8
client/timeProvider.cpp
View file

@ -1,16 +1,16 @@
#include "timeProvider.h" #include "timeProvider.h"
TimeProvider::TimeProvider() : diffToServer(0) TimeProvider::TimeProvider() : diffToServer_(0)
{ {
diffBuffer.setSize(200); diffBuffer_.setSize(200);
} }
void TimeProvider::setDiffToServer(double ms) void TimeProvider::setDiffToServer(double ms)
{ {
diffBuffer.add(ms * 1000); diffBuffer_.add(ms * 1000);
diffToServer = diffBuffer.median(); diffToServer_ = diffBuffer_.median();
} }
/* /*

6
client/timeProvider.h
View file

@ -22,7 +22,7 @@ public:
template<typename T> template<typename T>
inline T getDiffToServer() const inline T getDiffToServer() const
{ {
return std::chrono::duration_cast<T>(chronos::usec(diffToServer)); return std::chrono::duration_cast<T>(chronos::usec(diffToServer_));
} }
/* chronos::usec::rep getDiffToServer(); /* chronos::usec::rep getDiffToServer();
@ -56,8 +56,8 @@ private:
TimeProvider(TimeProvider const&); // Don't Implement TimeProvider(TimeProvider const&); // Don't Implement
void operator=(TimeProvider const&); // Don't implement void operator=(TimeProvider const&); // Don't implement
DoubleBuffer<chronos::usec::rep> diffBuffer; DoubleBuffer<chronos::usec::rep> diffBuffer_;
std::atomic<chronos::usec::rep> diffToServer; std::atomic<chronos::usec::rep> diffToServer_;
}; };

30
server/controlServer.cpp
View file

@ -7,7 +7,7 @@
#include <iostream> #include <iostream>
ControlServer::ControlServer(unsigned short port) : port_(port), headerChunk(NULL), sampleFormat(NULL) ControlServer::ControlServer(unsigned short port) : port_(port), headerChunk_(NULL), sampleFormat_(NULL)
{ {
} }
@ -17,7 +17,7 @@ ControlServer::ControlServer(unsigned short port) : port_(port), headerChunk(NUL
void ControlServer::send(shared_ptr<msg::BaseMessage> message) void ControlServer::send(shared_ptr<msg::BaseMessage> message)
{ {
std::unique_lock<std::mutex> mlock(mutex); std::unique_lock<std::mutex> mlock(mutex);
for (auto it = sessions.begin(); it != sessions.end(); ) for (auto it = sessions_.begin(); it != sessions_.end(); )
{ {
if (!(*it)->active()) if (!(*it)->active())
{ {
@ -25,13 +25,13 @@ void ControlServer::send(shared_ptr<msg::BaseMessage> message)
auto func = [](shared_ptr<ServerSession> s)->void{s->stop();}; auto func = [](shared_ptr<ServerSession> s)->void{s->stop();};
std::thread t(func, *it); std::thread t(func, *it);
t.detach(); t.detach();
sessions.erase(it++); sessions_.erase(it++);
} }
else else
++it; ++it;
} }
for (auto s : sessions) for (auto s : sessions_)
s->add(message); s->add(message);
} }
@ -56,18 +56,18 @@ void ControlServer::onMessageReceived(ServerSession* connection, const msg::Base
} }
else if (requestMsg.request == kServerSettings) else if (requestMsg.request == kServerSettings)
{ {
serverSettings->refersTo = requestMsg.id; serverSettings_->refersTo = requestMsg.id;
connection->send(serverSettings); connection->send(serverSettings_);
} }
else if (requestMsg.request == kSampleFormat) else if (requestMsg.request == kSampleFormat)
{ {
sampleFormat->refersTo = requestMsg.id; sampleFormat_->refersTo = requestMsg.id;
connection->send(sampleFormat); connection->send(sampleFormat_);
} }
else if (requestMsg.request == kHeader) else if (requestMsg.request == kHeader)
{ {
headerChunk->refersTo = requestMsg.id; headerChunk_->refersTo = requestMsg.id;
connection->send(headerChunk); connection->send(headerChunk_);
} }
} }
else if (baseMessage.type == message_type::kCommand) else if (baseMessage.type == message_type::kCommand)
@ -102,7 +102,7 @@ void ControlServer::acceptor()
{ {
std::unique_lock<std::mutex> mlock(mutex); std::unique_lock<std::mutex> mlock(mutex);
session->start(); session->start();
sessions.insert(shared_ptr<ServerSession>(session)); sessions_.insert(shared_ptr<ServerSession>(session));
} }
} }
} }
@ -110,7 +110,7 @@ void ControlServer::acceptor()
void ControlServer::start() void ControlServer::start()
{ {
acceptThread = new thread(&ControlServer::acceptor, this); acceptThread_ = new thread(&ControlServer::acceptor, this);
} }
@ -123,14 +123,14 @@ void ControlServer::stop()
void ControlServer::setHeader(msg::Header* header) void ControlServer::setHeader(msg::Header* header)
{ {
if (header) if (header)
headerChunk = header; headerChunk_ = header;
} }
void ControlServer::setFormat(msg::SampleFormat* format) void ControlServer::setFormat(msg::SampleFormat* format)
{ {
if (format) if (format)
sampleFormat = format; sampleFormat_ = format;
} }
@ -138,7 +138,7 @@ void ControlServer::setFormat(msg::SampleFormat* format)
void ControlServer::setServerSettings(msg::ServerSettings* settings) void ControlServer::setServerSettings(msg::ServerSettings* settings)
{ {
if (settings) if (settings)
serverSettings = settings; serverSettings_ = settings;
} }

14
server/controlServer.h
View file

@ -37,15 +37,15 @@ public:
private: private:
void acceptor(); void acceptor();
mutable std::mutex mutex; mutable std::mutex mutex_;
set<shared_ptr<ServerSession>> sessions; set<shared_ptr<ServerSession>> sessions_;
boost::asio::io_service io_service_; boost::asio::io_service io_service_;
unsigned short port_; unsigned short port_;
msg::Header* headerChunk; msg::Header* headerChunk_;
msg::SampleFormat* sampleFormat; msg::SampleFormat* sampleFormat_;
msg::ServerSettings* serverSettings; msg::ServerSettings* serverSettings_;
thread* acceptThread; thread* acceptThread_;
Queue<shared_ptr<msg::BaseMessage>> messages; Queue<shared_ptr<msg::BaseMessage>> messages_;
}; };

52
server/serverSession.cpp
View file

@ -9,9 +9,9 @@ using namespace std;
ServerSession::ServerSession(MessageReceiver* _receiver, std::shared_ptr<tcp::socket> _socket) : messageReceiver(_receiver) ServerSession::ServerSession(MessageReceiver* receiver, std::shared_ptr<tcp::socket> socket) : messageReceiver_(receiver)
{ {
socket = _socket; socket_ = socket;
} }
@ -24,9 +24,9 @@ ServerSession::~ServerSession()
void ServerSession::start() void ServerSession::start()
{ {
active_ = true; active_ = true;
streamActive = false; streamActive_ = false;
readerThread = new thread(&ServerSession::reader, this); readerThread_ = new thread(&ServerSession::reader, this);
writerThread = new thread(&ServerSession::writer, this); writerThread_ = new thread(&ServerSession::writer, this);
} }
@ -36,31 +36,31 @@ void ServerSession::stop()
try try
{ {
boost::system::error_code ec; boost::system::error_code ec;
if (socket) if (socket_)
{ {
socket->shutdown(boost::asio::ip::tcp::socket::shutdown_both, ec); socket_->shutdown(boost::asio::ip::tcp::socket::shutdown_both, ec);
if (ec) logE << "Error in socket shutdown: " << ec << "\n"; if (ec) logE << "Error in socket shutdown: " << ec << "\n";
socket->close(ec); socket_->close(ec);
if (ec) logE << "Error in socket close: " << ec << "\n"; if (ec) logE << "Error in socket close: " << ec << "\n";
} }
if (readerThread) if (readerThread_)
{ {
logD << "joining readerThread\n"; logD << "joining readerThread\n";
readerThread->join(); readerThread_->join();
delete readerThread; delete readerThread_;
} }
if (writerThread) if (writerThread_)
{ {
logD << "joining readerThread\n"; logD << "joining readerThread\n";
writerThread->join(); writerThread_->join();
delete writerThread; delete writerThread_;
} }
} }
catch(...) catch(...)
{ {
} }
readerThread = NULL; readerThread_ = NULL;
writerThread = NULL; writerThread_ = NULL;
logD << "ServerSession stopped\n"; logD << "ServerSession stopped\n";
} }
@ -72,7 +72,7 @@ void ServerSession::socketRead(void* _to, size_t _bytes)
do do
{ {
boost::system::error_code error; boost::system::error_code error;
read += socket->read_some(boost::asio::buffer((char*)_to + read, _bytes - read)); read += socket_->read_some(boost::asio::buffer((char*)_to + read, _bytes - read));
} }
while (read < _bytes); while (read < _bytes);
} }
@ -80,26 +80,26 @@ void ServerSession::socketRead(void* _to, size_t _bytes)
void ServerSession::add(shared_ptr<msg::BaseMessage> message) void ServerSession::add(shared_ptr<msg::BaseMessage> message)
{ {
if (!message || !streamActive) if (!message || !streamActive_)
return; return;
while (messages.size() > 100)// chunk->getDuration() > 10000) while (messages_.size() > 100)// chunk->getDuration() > 10000)
messages.pop(); messages_.pop();
messages.push(message); messages_.push(message);
} }
bool ServerSession::send(msg::BaseMessage* message) bool ServerSession::send(msg::BaseMessage* message)
{ {
std::unique_lock<std::mutex> mlock(mutex_); std::unique_lock<std::mutex> mlock(mutex_);
if (!socket) if (!socket_)
return false; return false;
boost::asio::streambuf streambuf; boost::asio::streambuf streambuf;
std::ostream stream(&streambuf); std::ostream stream(&streambuf);
tv t; tv t;
message->sent = t; message->sent = t;
message->serialize(stream); message->serialize(stream);
boost::asio::write(*socket.get(), streambuf); boost::asio::write(*socket_.get(), streambuf);
return true; return true;
} }
@ -119,8 +119,8 @@ void ServerSession::getNextMessage()
tv t; tv t;
baseMessage.received = t; baseMessage.received = t;
if (messageReceiver != NULL) if (messageReceiver_ != NULL)
messageReceiver->onMessageReceived(this, baseMessage, &buffer[0]); messageReceiver_->onMessageReceived(this, baseMessage, &buffer[0]);
} }
@ -154,7 +154,7 @@ void ServerSession::writer()
shared_ptr<msg::BaseMessage> message; shared_ptr<msg::BaseMessage> message;
while (active_) while (active_)
{ {
if (messages.try_pop(message, std::chrono::milliseconds(500))) if (messages_.try_pop(message, std::chrono::milliseconds(500)))
send(message.get()); send(message.get());
} }
} }

16
server/serverSession.h
View file

@ -28,7 +28,7 @@ public:
class ServerSession class ServerSession
{ {
public: public:
ServerSession(MessageReceiver* _receiver, std::shared_ptr<tcp::socket> _socket); ServerSession(MessageReceiver* receiver, std::shared_ptr<tcp::socket> socket);
~ServerSession(); ~ServerSession();
void start(); void start();
void stop(); void stop();
@ -42,7 +42,7 @@ public:
virtual void setStreamActive(bool active) virtual void setStreamActive(bool active)
{ {
streamActive = active; streamActive_ = active;
} }
@ -53,13 +53,13 @@ protected:
void writer(); void writer();
std::atomic<bool> active_; std::atomic<bool> active_;
std::atomic<bool> streamActive; std::atomic<bool> streamActive_;
mutable std::mutex mutex_; mutable std::mutex mutex_;
std::thread* readerThread; std::thread* readerThread_;
std::thread* writerThread; std::thread* writerThread_;
std::shared_ptr<tcp::socket> socket; std::shared_ptr<tcp::socket> socket_;
MessageReceiver* messageReceiver; MessageReceiver* messageReceiver_;
Queue<std::shared_ptr<msg::BaseMessage>> messages; Queue<std::shared_ptr<msg::BaseMessage>> messages_;
}; };