support pipeline multiple starts (#66)

Co-authored-by: zhupengdong <[email protected]>
ShawnNew · Jul 28, 2020 · d73ee1e · d73ee1e
1 parent 63d202d
commit d73ee1e
Show file tree

Hide file tree

Showing 4 changed files with 12 additions and 10 deletions.
diff --git a/framework/core/include/cnstream_pipeline.hpp b/framework/core/include/cnstream_pipeline.hpp
@@ -532,7 +532,6 @@ class Pipeline {
   std::unordered_map<std::string, ModuleAssociatedInfo> modules_;
   std::unordered_map<std::string, CNModuleConfig> modules_config_;
   std::unordered_map<std::string, std::vector<std::string>> connections_config_;
-  std::mutex stop_mtx_;
   uint64_t eos_mask_ = 0;
 
   std::vector<std::string> stream_ids_;

diff --git a/framework/core/src/cnstream_pipeline.cpp b/framework/core/src/cnstream_pipeline.cpp
@@ -324,6 +324,8 @@ bool Pipeline::QueryLinkStatus(LinkStatus* status, const std::string& link_id) {
 }
 
 bool Pipeline::Start() {
+  if (IsRunning()) return true;
+
   // set eos mask
   SetEOSMask();
   // open modules
@@ -391,7 +393,6 @@ bool Pipeline::Start() {
 }
 
 bool Pipeline::Stop() {
-  std::lock_guard<std::mutex> lk(stop_mtx_);
   if (!IsRunning()) return true;
 
   // stop data transmit

diff --git a/modules/source/src/data_handler_rtsp.cpp b/modules/source/src/data_handler_rtsp.cpp
@@ -363,10 +363,6 @@ bool RtspHandlerImpl::Open() {
   // updated with paramSet
   DataSource *source = dynamic_cast<DataSource *>(module_);
   param_ = source->GetSourceParam();
-  if (param_.output_type_ != OUTPUT_MLU) {
-    LOG(ERROR) << "output_type not supported:" << param_.output_type_;
-    return false;
-  }
   this->interval_ = param_.interval_;
 
   SetPerfManager(source->GetPerfManager(stream_id_));

diff --git a/samples/demo/demo.cpp b/samples/demo/demo.cpp
@@ -86,7 +86,9 @@ class MsgObserver : cnstream::StreamMsgObserver {
       LOG(ERROR) << "[Observer] received ERROR_MSG";
       stop_ = true;
     }
-    wakener_.notify_one();
+    if (stop_) {
+      wakener_.notify_one();
+    }
   }
 
   void WaitForStop() {
@@ -98,6 +100,7 @@ class MsgObserver : cnstream::StreamMsgObserver {
 
   void IncreaseStreamCnt() { stream_cnt_++; }
   void DecreaseStreamCnt() { stream_cnt_--; }
+  int GetStreamCnt() { return stream_cnt_.load(); }
 
  private:
   std::atomic<int> stream_cnt_;
@@ -294,9 +297,6 @@ int main(int argc, char** argv) {
   cnstream::DataSource* source = dynamic_cast<cnstream::DataSource*>(pipeline.GetModule("source"));
 #ifdef BUILD_IPC
   cnstream::ModuleIPC* ipc = dynamic_cast<cnstream::ModuleIPC*>(pipeline.GetModule("ipc"));
-  if (ipc != nullptr) {
-    ipc->SetChannelCount(video_urls.size());
-  }
   if (nullptr == source && (nullptr == ipc)) {
     LOG(ERROR) << "DataSource && ModuleIPC module both not found.";
 #else
@@ -357,6 +357,12 @@ int main(int argc, char** argv) {
     }
   }
 
+#ifdef BUILD_IPC
+  if (nullptr != ipc) {
+    ipc->SetChannelCount(msg_observer.GetStreamCnt());
+  }
+#endif
+
   auto quit_callback = [&pipeline, streams, &source]() {
     // stop feed-data threads before remove-sources...
     thread_running.store(false);