report stream state

server/config.cpp

@@ -55,7 +55,7 @@ Config::Config()
 				client->fromJson(*it);
 				client->connected = false;
 				clients.push_back(client);
-				std::cout << "Client:\n" << std::setw(4) << client->toJson() << '\n';
+//				logO << "Client:\n" << std::setw(4) << client->toJson() << '\n';
 			}
 		}
 	}

server/pcmreader/fileReader.cpp

@@ -58,6 +58,8 @@ void FileReader::worker()
 	size_t length = ifs.tellg();
 	ifs.seekg (0, ifs.beg);
 
+	setState(kPlaying);
+
 	while (active_)
 	{
 		gettimeofday(&tvChunk, NULL);

server/pcmreader/pcmReader.cpp

@@ -32,7 +32,7 @@ using namespace std;
 
 
 
-PcmReader::PcmReader(PcmListener* pcmListener, const ReaderUri& uri) : pcmListener_(pcmListener), uri_(uri), pcmReadMs_(20)
+PcmReader::PcmReader(PcmListener* pcmListener, const ReaderUri& uri) : pcmListener_(pcmListener), uri_(uri), pcmReadMs_(20), state_(kIdle)
 {
 	EncoderFactory encoderFactory;
  	if (uri_.query.find("codec") == uri_.query.end())
@@ -104,6 +104,22 @@ void PcmReader::stop()
 }
 
 
+ReaderState PcmReader::getState() const
+{
+	return state_;
+}
+
+
+void PcmReader::setState(const ReaderState& newState)
+{
+	if (newState != state_)
+	{
+		state_ = newState;
+		pcmListener_->onStateChanged(this, newState);
+	}
+}
+
+
 void PcmReader::onChunkEncoded(const Encoder* encoder, msg::PcmChunk* chunk, double duration)
 {
 //	logO << "onChunkEncoded: " << duration << " us\n";
@@ -116,3 +132,22 @@ void PcmReader::onChunkEncoded(const Encoder* encoder, msg::PcmChunk* chunk, dou
 	pcmListener_->onChunkRead(this, chunk, duration);
 }
 
+
+json PcmReader::toJson() const
+{
+	string state("idle");
+	if (state_ == kIdle)
+		state = "idle";
+	else if (state_ == kPlaying)
+		state = "playing";
+	else if (state_ == kDisabled)
+		state = "disabled";
+
+	json j = {
+		{"uri", uri_.toJson()},
+		{"id", uri_.id()},
+		{"status", state}
+	};
+	return j;
+}
+

server/pcmreader/pcmReader.h

@@ -25,12 +25,20 @@
 #include <map>
 #include "readerUri.h"
 #include "../encoder/encoder.h"
+#include "../json/json.hpp"
 #include "message/sampleFormat.h"
 #include "message/header.h"
 
 
 class PcmReader;
 
+enum ReaderState
+{
+	kIdle = 0,
+	kPlaying = 1,
+	kDisabled = 2
+};
+
 
 /// Callback interface for users of PcmReader
 /**
@@ -39,6 +47,7 @@ class PcmReader;
 class PcmListener
 {
 public:
+	virtual void onStateChanged(const PcmReader* pcmReader, const ReaderState& state) = 0;
 	virtual void onChunkRead(const PcmReader* pcmReader, const msg::PcmChunk* chunk, double duration) = 0;
 	virtual void onResync(const PcmReader* pcmReader, double ms) = 0;
 };
@@ -68,8 +77,14 @@ public:
 	virtual const std::string& getName() const;
 	virtual const SampleFormat& getSampleFormat() const;
 
+	virtual ReaderState getState() const;
+	virtual json toJson() const;
+
+
 protected:
 	virtual void worker() = 0;
+	void setState(const ReaderState& newState);
+
 	timeval tvEncodedChunk_;
 	std::atomic<bool> active_;
 	std::thread readerThread_;
@@ -79,6 +94,7 @@ protected:
 	size_t pcmReadMs_;
 	std::unique_ptr<Encoder> encoder_;
 	std::string name_;
+	ReaderState state_;
 };
 
 

server/pcmreader/pipeReader.cpp

@@ -76,7 +76,10 @@ void PipeReader::worker()
 				{
 					int count = read(fd_, chunk->payload + len, toRead - len);
 					if (count < 0)
+					{
+						setState(kIdle);
 						usleep(100*1000);
+					}
 					else if (count == 0)
 						throw SnapException("end of file");
 					else
@@ -92,6 +95,7 @@ void PipeReader::worker()
 				if (nextTick >= currentTick)
 				{
 //					logO << "sleep: " << nextTick - currentTick << "\n";
+					setState(kPlaying);
 					usleep((nextTick - currentTick) * 1000);
 				}
 				else

server/pcmreader/readerUri.cpp

@@ -108,14 +108,14 @@ ReaderUri::ReaderUri(const std::string& readerUri)
 
 json ReaderUri::toJson() const
 {
-	json j;
-	j["uri"] = uri;
-	j["scheme"] = scheme;
-	j["host"] = host;
-	j["path"] = path;
-	j["fragment"] = fragment;
-	j["query"] = json(query);
-	j["id"] = id_;
+	json j = {
+		{"raw", uri},
+		{"scheme", scheme},
+		{"host", host},
+		{"path", path},
+		{"fragment", fragment},
+		{"query", query}
+	};
 	return j;
 }
 

server/pcmreader/streamManager.cpp

@@ -114,7 +114,7 @@ json StreamManager::toJson() const
 {
 	json result = json::array();
 	for (auto stream: streams_)
-		result.push_back(stream->getUri().toJson());
+		result.push_back(stream->toJson());
 	return result;
 }
 

server/streamServer.cpp

@@ -40,6 +40,15 @@ StreamServer::~StreamServer()
 }
 
 
+void StreamServer::onStateChanged(const PcmReader* pcmReader, const ReaderState& state)
+{
+	logO << "onStateChanged (" << pcmReader->getName() << "): " << state << "\n";
+//	logO << pcmReader->toJson().dump(4);
+	json notification = JsonNotification::getJson("Stream.OnUpdate", pcmReader->toJson());
+	controlServer_->send(notification.dump(), NULL);
+}
+
+
 void StreamServer::onChunkRead(const PcmReader* pcmReader, const msg::PcmChunk* chunk, double duration)
 {
 //	logO << "onChunkRead (" << pcmReader->getName() << "): " << duration << "ms\n";
@@ -301,7 +310,7 @@ void StreamServer::onMessageReceived(StreamSession* connection, const msg::BaseM
 		connection->send(headerChunk.get());
 
 		json notification = JsonNotification::getJson("Client.OnConnect", client->toJson());
-		logO << notification.dump(4) << "\n";
+//		logO << notification.dump(4) << "\n";
 		controlServer_->send(notification.dump());
 	}
 }

server/streamServer.h

@@ -89,6 +89,7 @@ public:
 	virtual void onMessageReceived(ControlSession* connection, const std::string& message);
 
 	/// Implementation of PcmListener
+	virtual void onStateChanged(const PcmReader* pcmReader, const ReaderState& state);
 	virtual void onChunkRead(const PcmReader* pcmReader, const msg::PcmChunk* chunk, double duration);
 	virtual void onResync(const PcmReader* pcmReader, double ms);