stream notifies about new chunks

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/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_;
 };
 
 

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