Reintroduce player state

server/streamreader/asio_stream.hpp

@@ -20,9 +20,11 @@
 #define ASIO_STREAM_HPP
 
 #include "pcm_stream.hpp"
+#include <atomic>
 #include <boost/asio.hpp>
 
 
+
 template <typename ReadStream>
 class AsioStream : public PcmStream, public std::enable_shared_from_this<AsioStream<ReadStream>>
 {
@@ -38,20 +40,45 @@ protected:
     virtual void disconnect() = 0;
     virtual void on_connect();
     virtual void do_read();
+    void check_state();
     std::unique_ptr<msg::PcmChunk> chunk_;
     timeval tv_chunk_;
     bool first_;
     long nextTick_;
     boost::asio::deadline_timer timer_;
+    boost::asio::deadline_timer idle_timer_;
     std::unique_ptr<ReadStream> stream_;
+    std::atomic<std::uint64_t> bytes_read_;
 };
 
 
 
 template <typename ReadStream>
-AsioStream<ReadStream>::AsioStream(PcmListener* pcmListener, boost::asio::io_context& ioc, const StreamUri& uri) : PcmStream(pcmListener, ioc, uri), timer_(ioc)
+AsioStream<ReadStream>::AsioStream(PcmListener* pcmListener, boost::asio::io_context& ioc, const StreamUri& uri)
+    : PcmStream(pcmListener, ioc, uri), timer_(ioc), idle_timer_(ioc)
 {
     chunk_ = std::make_unique<msg::PcmChunk>(sampleFormat_, pcmReadMs_);
+    bytes_read_ = 0;
+}
+
+
+template <typename ReadStream>
+void AsioStream<ReadStream>::check_state()
+{
+    uint64_t last_read = bytes_read_;
+    auto self = this->shared_from_this();
+    idle_timer_.expires_from_now(boost::posix_time::milliseconds(500 + pcmReadMs_));
+    idle_timer_.async_wait([self, this, last_read](const boost::system::error_code& ec) {
+        if (!ec)
+        {
+            LOG(DEBUG) << "check state last: " << last_read << ", read: " << bytes_read_ << "\n";
+            if (bytes_read_ != last_read)
+                setState(kPlaying);
+            else
+                setState(kIdle);
+            check_state();
+        }
+    });
 }
 
 
@@ -59,6 +86,7 @@ template <typename ReadStream>
 void AsioStream<ReadStream>::start()
 {
     encoder_->init(this, sampleFormat_);
+    check_state();
     connect();
 }
 
@@ -67,6 +95,7 @@ template <typename ReadStream>
 void AsioStream<ReadStream>::stop()
 {
     timer_.cancel();
+    idle_timer_.cancel();
     disconnect();
 }
 
@@ -97,6 +126,8 @@ void AsioStream<ReadStream>::do_read()
                                     connect();
                                     return;
                                 }
+
+                                bytes_read_ += length;
                                 // LOG(DEBUG) << "Read: " << length << " bytes\n";
                                 if (first_)
                                 {
@@ -114,7 +145,6 @@ void AsioStream<ReadStream>::do_read()
 
                                 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)

server/streamreader/pcm_stream.cpp

@@ -138,6 +138,7 @@ void PcmStream::setState(const ReaderState& newState)
 {
     if (newState != state_)
     {
+        LOG(DEBUG) << "State changed: " << state_ << " => " << newState << "\n";
         state_ = newState;
         if (pcmListener_)
             pcmListener_->onStateChanged(this, newState);

server/streamreader/pipe_stream.cpp

@@ -25,7 +25,6 @@
 #include "common/aixlog.hpp"
 #include "common/snap_exception.hpp"
 #include "common/str_compat.hpp"
-#include "encoder/encoder_factory.hpp"
 #include "pipe_stream.hpp"
 
 
@@ -33,8 +32,7 @@ using namespace std;
 
 
 
-PipeStream::PipeStream(PcmListener* pcmListener, boost::asio::io_context& ioc, const StreamUri& uri)
+PipeStream::PipeStream(PcmListener* pcmListener, boost::asio::io_context& ioc, const StreamUri& uri) : AsioStream<stream_descriptor>(pcmListener, ioc, uri)
-    : AsioStream<boost::asio::posix::stream_descriptor>(pcmListener, ioc, uri)
 {
     umask(0);
     string mode = uri_.getQuery("mode", "create");