thread skeleton

src/common/AdaptiveStream.cpp

@@ -32,9 +32,16 @@ AdaptiveStream::AdaptiveStream(AdaptiveTree &tree, AdaptiveTree::StreamType type
   , current_period_(tree_.periods_.empty() ? 0 : tree_.periods_[0])
   , current_adp_(0)
   , current_rep_(0)
+  , thread_data_(nullptr)
 {
 }
 
+AdaptiveStream::~AdaptiveStream()
+{
+  stop();
+  clear();
+}
+
 bool AdaptiveStream::download_segment()
 {
   segment_buffer_.clear();
@@ -115,6 +122,15 @@ bool AdaptiveStream::download_segment()
   return false;
 }
 
+void AdaptiveStream::worker()
+{
+  do {
+    std::unique_lock<std::mutex> lck(thread_data_->mutex_dl_);
+    thread_data_->signal_dl_.wait(lck);
+  } while (!thread_data_->thread_stop_);
+}
+
+
 bool AdaptiveStream::write_data(const void *buffer, size_t buffer_size)
 {
   segment_buffer_ += std::string((const char *)buffer, buffer_size);
@@ -190,6 +206,13 @@ bool AdaptiveStream::start_stream(const uint32_t seg_offset, uint16_t width, uin
 
     stopped_ = false;
   }
+
+  if (!thread_data_ && !stopped_)
+  {
+    thread_data_ = new THREADDATA();
+    thread_data_->Start(this);
+  }
+
   return true;
 }
 
@@ -392,13 +415,19 @@ void AdaptiveStream::info(std::ostream &s)
   s << ts[type_] << " representation: " << current_rep_->url_.substr(current_rep_->url_.find_last_of('/') + 1) << " bandwidth: " << current_rep_->bandwidth_ << std::endl;
 }
 
+void AdaptiveStream::stop()
+{
+  stopped_ = true;
+  if (thread_data_)
+  {
+    delete thread_data_;
+    thread_data_ = nullptr;
+  }
+};
+
 void AdaptiveStream::clear()
 {
   current_adp_ = 0;
   current_rep_ = 0;
-}
 
-AdaptiveStream::~AdaptiveStream()
-{
-  clear();
 }

src/common/AdaptiveStream.h

@@ -22,6 +22,9 @@
 #include <string>
 #include <map>
 
+#include <thread>
+#include <mutex>
+
 namespace adaptive
 {
   class AdaptiveStream;
@@ -37,7 +40,7 @@ namespace adaptive
   {
   public:
     AdaptiveStream(AdaptiveTree &tree, AdaptiveTree::StreamType type);
-    ~AdaptiveStream();
+    virtual ~AdaptiveStream();
     void set_observer(AdaptiveStreamObserver *observer){ observer_ = observer; };
     bool prepare_stream(const AdaptiveTree::AdaptationSet *adp,
       const uint32_t width, const uint32_t height, uint32_t hdcpLimit, uint16_t hdcpVersion,
@@ -46,7 +49,7 @@ namespace adaptive
     bool start_stream(const uint32_t seg_offset, uint16_t width, uint16_t height);
     bool restart_stream();
     bool select_stream(bool force = false, bool justInit = false, unsigned int repId = 0);
-    void stop(){ stopped_ = true; };
+    void stop();
     void clear();
     void info(std::ostream &s);
     unsigned int getWidth() const { return width_; };
@@ -73,7 +76,35 @@ namespace adaptive
     virtual bool parseIndexRange() { return false; };
     bool write_data(const void *buffer, size_t buffer_size);
   private:
+    // Segment download section
     bool download_segment();
+    void worker();
+
+    struct THREADDATA
+    {
+      THREADDATA()
+        : thread_stop_(false)
+      {
+      }
+
+      void Start(AdaptiveStream *parent)
+      {
+        download_thread_ = std::thread(&AdaptiveStream::worker, parent);
+      }
+
+      ~THREADDATA()
+      {
+        thread_stop_ = true;
+        signal_dl_.notify_one();
+        download_thread_.join();
+      };
+
+      std::mutex mutex_rw_, mutex_dl_;
+      std::condition_variable signal_rw_, signal_dl_;
+      std::thread download_thread_;
+      bool thread_stop_;
+    };
+    THREADDATA *thread_data_;
 
     AdaptiveTree &tree_;
     AdaptiveTree::StreamType type_;