merged develop branch

client/Makefile

@@ -1,4 +1,4 @@
-VERSION = 0.3.0
+VERSION = 0.3.1
 TARGET  = snapclient
 SHELL = /bin/bash
 

client/alsaPlayer.cpp

@@ -201,13 +201,15 @@ void Player::worker()
 		else
 		{
 			logO << "Failed to get chunk\n";
-			usleep(100*1000);
+			while (active_ && !stream_->waitForChunk(100))
+				logD << "Waiting for chunk\n";
 		}
 	}
 }
 
 
-vector<PcmDevice> Player::pcm_list(void) {
+vector<PcmDevice> Player::pcm_list(void)
+{
 	void **hints, **n;
 	char *name, *descr, *io;
 	vector<PcmDevice> result;

client/controller.cpp

@@ -36,7 +36,7 @@
 using namespace std;
 
 
-Controller::Controller() : MessageReceiver(), active_(false), sampleFormat_(NULL), decoder_(NULL)
+Controller::Controller() : MessageReceiver(), active_(false), sampleFormat_(NULL), decoder_(NULL), sendTimeSyncMsg_(false), asyncException_(false)
 {
 }
 
@@ -44,6 +44,8 @@ Controller::Controller() : MessageReceiver(), active_(false), sampleFormat_(NULL
 void Controller::onException(ClientConnection* connection, const std::exception& exception)
 {
 	logE << "onException: " << exception.what() << "\n";
+	exception_ = exception;
+	asyncException_ = true;
 }
 
 
@@ -66,6 +68,22 @@ void Controller::onMessageReceived(ClientConnection* connection, const msg::Base
 				delete pcmChunk;
 		}
 	}
+	else if (baseMessage.type == message_type::kTime)
+	{
+		msg::Time reply;
+		reply.deserialize(baseMessage, buffer);
+		double latency = (reply.received.sec - reply.sent.sec) + (reply.received.usec - reply.sent.usec) / 1000000.;
+//		logO << "diff to server [ms]: " << (float)TimeProvider::getInstance().getDiffToServer<chronos::usec>().count() / 1000.f << "\n";
+//		logO << "timeMsg: " << latency << "\n";
+		TimeProvider::getInstance().setDiffToServer((reply.latency - latency) * 1000 / 2);
+	}
+
+	if (sendTimeSyncMsg_)
+	{
+		msg::Request timeReq(kTime);
+		clientConnection_->send(&timeReq);
+		sendTimeSyncMsg_ = false;
+	}
 }
 
 
@@ -147,17 +165,25 @@ void Controller::worker()
 
 			while (active_)
 			{
-				usleep(500*1000);
-                shared_ptr<msg::Time> reply = clientConnection_->sendReq<msg::Time>(&timeReq);
-                if (reply)
-                {
-                    double latency = (reply->received.sec - reply->sent.sec) + (reply->received.usec - reply->sent.usec) / 1000000.;
-                    TimeProvider::getInstance().setDiffToServer((reply->latency - latency) * 1000 / 2);
-                }
+				for (size_t n=0; n<10 && active_; ++n)
+				{
+					usleep(100*1000);
+					if (asyncException_)
+						throw exception_;
+				}
+
+				sendTimeSyncMsg_ = true;
+//				shared_ptr<msg::Time> reply = clientConnection_->sendReq<msg::Time>(&timeReq);
+//				if (reply)
+//				{
+//					double latency = (reply->received.sec - reply->sent.sec) + (reply->received.usec - reply->sent.usec) / 1000000.;
+//					TimeProvider::getInstance().setDiffToServer((reply->latency - latency) * 1000 / 2);
+//				}
 			}
 		}
 		catch (const std::exception& e)
 		{
+			asyncException_ = false;
 			logS(kLogErr) << "Exception in Controller::worker(): " << e.what() << endl;
 			logO << "Stopping clientConnection" << endl;
 			clientConnection_->stop();

client/controller.h

@@ -60,6 +60,9 @@ private:
 	Decoder* decoder_;
 	PcmDevice pcmDevice_;
 	size_t latency_;
+	bool sendTimeSyncMsg_;
+	std::exception exception_;
+	bool asyncException_;
 };
 
 

client/stream.cpp

@@ -74,10 +74,20 @@ void Stream::addChunk(msg::PcmChunk* chunk)
 	while (chunks_.size() * chunk->duration<cs::msec>().count() > 10000)
 		chunks_.pop();
 	chunks_.push(shared_ptr<msg::PcmChunk>(chunk));
+	std::unique_lock<std::mutex> lck(cvMutex_);
+	cv_.notify_one();
 //	logD << "new chunk: " << chunk_->getDuration() << ", Chunks: " << chunks_.size() << "\n";
 }
 
 
+bool Stream::waitForChunk(size_t ms) const
+{
+	std::unique_lock<std::mutex> lck(cvMutex_);
+	cv_.wait_for(lck, std::chrono::milliseconds(ms));
+	return !chunks_.empty();
+}
+
+
 
 cs::time_point_hrc Stream::getSilentPlayerChunk(void* outputBuffer, unsigned long framesPerBuffer)
 {

client/stream.h

@@ -54,12 +54,14 @@ public:
 
 	/// "Server buffer": playout latency, e.g. 1000ms
 	void setBufferLen(size_t bufferLenMs);
-	
+
 	const msg::SampleFormat& getFormat() const
 	{
 		return format_;
 	}
 
+	bool waitForChunk(size_t ms) const;
+
 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, long framesCorrection);
@@ -88,6 +90,9 @@ private:
 	unsigned long playedFrames_;
 	long correctAfterXFrames_;
 	chronos::msec bufferMs_;
+
+	mutable std::condition_variable cv_;
+	mutable std::mutex cvMutex_;
 };
 
 

client/timeProvider.cpp

@@ -17,9 +17,10 @@
 ***/
 
 #include "timeProvider.h"
+#include "common/log.h"
 
 
-TimeProvider::TimeProvider() : diffToServer_(0)
+TimeProvider::TimeProvider() : diffToServer_(0), lastTimeSync_(0)
 {
 	diffBuffer_.setSize(200);
 }
@@ -27,8 +28,19 @@ TimeProvider::TimeProvider() : diffToServer_(0)
 
 void TimeProvider::setDiffToServer(double ms)
 {
-	diffBuffer_.add(ms * 1000);
+	long now = chronos::getTickCount();
+	/// clear diffBuffer if last update is older than a minute
+	if (now > lastTimeSync_ + 60*1000)
+	{
+		logO << "Last time sync older than a minute. Clearing time buffer\n";
+		diffToServer_ = ms*1000;
+		diffBuffer_.clear();
+	}
+	lastTimeSync_ = now;
+
+	diffBuffer_.add(ms*1000);
 	diffToServer_ = diffBuffer_.median();
+//	logO << "setDiffToServer: " << ms << ", diff: " << diffToServer_ / 1000.f << "\n";
 }
 
 /*

client/timeProvider.h

@@ -82,6 +82,7 @@ private:
 
 	DoubleBuffer<chronos::usec::rep> diffBuffer_;
 	std::atomic<chronos::usec::rep> diffToServer_;
+	long lastTimeSync_;
 };
 
 

common/queue.h

@@ -101,6 +101,11 @@ public:
 		return queue_.size();
 	}
 
+	bool empty() const
+	{
+		return (size() == 0);
+	}
+
 	Queue()=default;
 	Queue(const Queue&) = delete;            // disable copying
 	Queue& operator=(const Queue&) = delete; // disable assignment

server/Makefile

@@ -1,4 +1,4 @@
-VERSION = 0.3.0
+VERSION = 0.3.1
 TARGET  = snapserver
 SHELL = /bin/bash
 

server/encoder/flacEncoder.cpp

@@ -58,7 +58,7 @@ std::string FlacEncoder::getAvailableOptions() const
 
 std::string FlacEncoder::getDefaultOptions() const
 {
-	return "5";
+	return "2";
 }