data_source_pase add parm session config

pansgg · CarlosLeeGit · commit 6fe140181c5b · 2022-09-23T09:25:02.000+08:00
diff --git a/src/drivers/common/flowunit/source_context/source_context.cc b/src/drivers/common/flowunit/source_context/source_context.cc
@@ -31,8 +31,8 @@ std::shared_ptr<std::string> SourceContext::GetSourceURL() {
   std::string uri_str;
   DestroyUriFunc destroy_uri_func;
 
-  auto ret = plugin_->Parse(session_context_, data_source_cfg_, uri_str,
-                            destroy_uri_func);
+  auto ret = plugin_->Parse(session_context_, session_config_, data_source_cfg_,
+                            uri_str, destroy_uri_func);
   if (!ret) {
     MBLOG_ERROR << "Parse config failed, source uri is empty";
     return nullptr;
@@ -100,6 +100,11 @@ void SourceContext::SetSessionContext(
   session_context_ = session_context;
 }
 
+void SourceContext::SetSessionConfig(
+    const std::shared_ptr<modelbox::Configuration> &session_config) {
+  session_config_ = session_config;
+}
+
 int32_t SourceContext::GetRetryInterval() {
   return retry_context_.GetRetryInterval();
 }
diff --git a/src/drivers/common/flowunit/source_context/source_context.h b/src/drivers/common/flowunit/source_context/source_context.h
@@ -71,6 +71,8 @@ class SourceContext {
   void SetDataSourceCfg(std::string data_source_cfg);
   void SetSessionContext(
       const std::shared_ptr<modelbox::SessionContext>& session_context);
+  void SetSessionConfig(
+      const std::shared_ptr<modelbox::Configuration>& session_config);
 
   RetryStatus NeedRetry();
   int32_t GetRetryInterval();
@@ -79,6 +81,7 @@ class SourceContext {
 
  private:
   std::shared_ptr<modelbox::SessionContext> session_context_;
+  std::shared_ptr<modelbox::Configuration> session_config_;
   std::string data_source_cfg_;
   std::shared_ptr<DataSourceParserPlugin> plugin_;
   modelbox::Status last_status_{modelbox::STATUS_SUCCESS};
diff --git a/src/drivers/devices/cpu/flowunit/data_source_parser/data_source_parser_flowunit.cc b/src/drivers/devices/cpu/flowunit/data_source_parser/data_source_parser_flowunit.cc
@@ -91,8 +91,9 @@ modelbox::Status DataSourceParserFlowUnit::Process(
              << session_ctx->GetSessionId();
   std::string data_source_cfg(inbuff_data, buffer->GetBytes());
   std::shared_ptr<std::string> uri;
+  auto session_config = data_ctx->GetSessionConfig();
   std::shared_ptr<modelbox::SourceContext> source_context =
-      Parse(session_ctx, source_type, data_source_cfg, uri);
+      Parse(session_ctx, session_config, source_type, data_source_cfg, uri);
   if (source_context) {
     source_context->SetDataSourceCfg(data_source_cfg);
   } else {
@@ -111,6 +112,7 @@ modelbox::Status DataSourceParserFlowUnit::Process(
 
 std::shared_ptr<modelbox::SourceContext> DataSourceParserFlowUnit::Parse(
     const std::shared_ptr<modelbox::SessionContext> &session_context,
+    const std::shared_ptr<modelbox::Configuration> &session_config,
     const std::string &source_type, const std::string &data_source_cfg,
     std::shared_ptr<std::string> &uri) {
   auto plugin = GetPlugin(source_type);
@@ -123,9 +125,8 @@ std::shared_ptr<modelbox::SourceContext> DataSourceParserFlowUnit::Parse(
   std::string uri_str;
   modelbox::DestroyUriFunc destroy_uri_func;
   std::string stream_type;
-
-  auto ret = plugin->Parse(session_context, data_source_cfg, uri_str,
-                           destroy_uri_func);
+  auto ret = plugin->Parse(session_context, session_config, data_source_cfg,
+                           uri_str, destroy_uri_func);
   if (!ret) {
     MBLOG_ERROR << "Parse config failed, source uri is empty";
   }
@@ -138,6 +139,7 @@ std::shared_ptr<modelbox::SourceContext> DataSourceParserFlowUnit::Parse(
   plugin->GetStreamType(data_source_cfg, stream_type);
   source_context->SetStreamType(stream_type);
   source_context->SetSessionContext(session_context);
+  source_context->SetSessionConfig(session_config);
 
   uri = std::shared_ptr<std::string>(new std::string(uri_str),
                                      [destroy_uri_func](std::string *ptr) {
diff --git a/src/drivers/devices/cpu/flowunit/data_source_parser/data_source_parser_flowunit.h b/src/drivers/devices/cpu/flowunit/data_source_parser/data_source_parser_flowunit.h
@@ -70,6 +70,7 @@ class DataSourceParserFlowUnit : public modelbox::FlowUnit {
  private:
   std::shared_ptr<modelbox::SourceContext> Parse(
       const std::shared_ptr<modelbox::SessionContext> &session_context,
+      const std::shared_ptr<modelbox::Configuration> &session_config,
       const std::string &source_type, const std::string &data_source_cfg,
       std::shared_ptr<std::string> &uri);
 
diff --git a/src/drivers/devices/cpu/flowunit/data_source_parser/parser_plugin/obs_source_parser/obs_source_parser.cc b/src/drivers/devices/cpu/flowunit/data_source_parser/parser_plugin/obs_source_parser/obs_source_parser.cc
@@ -55,6 +55,11 @@ modelbox::Status ObsSourceParser::Init(
   retry_interval_ = opts->GetInt32("obs_retry_interval_ms", retry_interval_);
   retry_max_times_ = opts->GetInt32("obs_retry_count_limit", retry_max_times_);
   read_type_ = opts->GetString("obs_download_method", "file");
+
+  MBLOG_INFO << "obs source parser config retry_enabled:" << retry_enabled_
+             << " retry_interval:" << retry_interval_
+             << " retry_max_times:" << retry_max_times_;
+
   if (read_type_ != "stream") {
     return modelbox::STATUS_OK;
   }
@@ -67,17 +72,14 @@ modelbox::Status ObsSourceParser::Init(
     max_read_size_ = OBS_STREAM_READ_SIZE_HIGH;
   }
 
-  MBLOG_INFO << "obs source parser config retry_enabled:" << retry_enabled_
-             << " retry_interval:" << retry_interval_
-             << " retry_max_times:" << retry_max_times_;
-
   return modelbox::STATUS_OK;
 }
 
 modelbox::Status ObsSourceParser::Deinit() { return modelbox::STATUS_OK; }
 
 modelbox::Status ObsSourceParser::Parse(
     const std::shared_ptr<modelbox::SessionContext> &session_context,
+    const std::shared_ptr<modelbox::Configuration> &session_config,
     const std::string &config, std::string &uri,
     modelbox::DestroyUriFunc &destroy_uri_func) {
   OBSDownloadInfo download_info;
diff --git a/src/drivers/devices/cpu/flowunit/data_source_parser/parser_plugin/obs_source_parser/obs_source_parser.h b/src/drivers/devices/cpu/flowunit/data_source_parser/parser_plugin/obs_source_parser/obs_source_parser.h
@@ -55,6 +55,7 @@ class ObsSourceParser : public modelbox::DataSourceParserPlugin {
 
   modelbox::Status Parse(
       const std::shared_ptr<modelbox::SessionContext> &session_context,
+      const std::shared_ptr<modelbox::Configuration> &session_config,
       const std::string &config, std::string &uri,
       modelbox::DestroyUriFunc &destroy_uri_func) override;
   modelbox::Status GetStreamType(const std::string &config,
diff --git a/src/drivers/devices/cpu/flowunit/data_source_parser/parser_plugin/restful_source_parser/restful_source_parser.cc b/src/drivers/devices/cpu/flowunit/data_source_parser/parser_plugin/restful_source_parser/restful_source_parser.cc
@@ -35,13 +35,19 @@ modelbox::Status RestfulSourceParser::Init(
       opts->GetInt32("restful_retry_interval_ms", retry_interval_);
   retry_max_times_ =
       opts->GetInt32("restful_retry_count_limit", retry_max_times_);
+
+  MBLOG_INFO << "restful source parser config retry_enabled:" << retry_enabled_
+             << " retry_interval:" << retry_interval_
+             << " retry_max_times:" << retry_max_times_;
+
   return modelbox::STATUS_OK;
 }
 
 modelbox::Status RestfulSourceParser::Deinit() { return modelbox::STATUS_OK; }
 
 modelbox::Status RestfulSourceParser::Parse(
     const std::shared_ptr<modelbox::SessionContext> &session_context,
+    const std::shared_ptr<modelbox::Configuration> &session_config,
     const std::string &config, std::string &uri,
     modelbox::DestroyUriFunc &destroy_uri_func) {
   RestfulInputInfo input_info;
diff --git a/src/drivers/devices/cpu/flowunit/data_source_parser/parser_plugin/restful_source_parser/restful_source_parser.h b/src/drivers/devices/cpu/flowunit/data_source_parser/parser_plugin/restful_source_parser/restful_source_parser.h
@@ -48,6 +48,7 @@ class RestfulSourceParser : public modelbox::DataSourceParserPlugin {
 
   modelbox::Status Parse(
       const std::shared_ptr<modelbox::SessionContext> &session_context,
+      const std::shared_ptr<modelbox::Configuration> &session_config,
       const std::string &config, std::string &uri,
       modelbox::DestroyUriFunc &destroy_uri_func) override;
 
diff --git a/src/drivers/devices/cpu/flowunit/data_source_parser/parser_plugin/url_source_parser/url_source_parser.cc b/src/drivers/devices/cpu/flowunit/data_source_parser/parser_plugin/url_source_parser/url_source_parser.cc
@@ -88,6 +88,7 @@ modelbox::Status UrlSourceParser::GetStreamType(const std::string &config,
 
 modelbox::Status UrlSourceParser::Parse(
     const std::shared_ptr<modelbox::SessionContext> &session_context,
+    const std::shared_ptr<modelbox::Configuration> &session_config,
     const std::string &config, std::string &uri,
     modelbox::DestroyUriFunc &destroy_uri_func) {
   nlohmann::json json;
diff --git a/src/drivers/devices/cpu/flowunit/data_source_parser/parser_plugin/url_source_parser/url_source_parser.h b/src/drivers/devices/cpu/flowunit/data_source_parser/parser_plugin/url_source_parser/url_source_parser.h
@@ -37,6 +37,7 @@ class UrlSourceParser : public modelbox::DataSourceParserPlugin {
 
   modelbox::Status Parse(
       const std::shared_ptr<modelbox::SessionContext> &session_context,
+      const std::shared_ptr<modelbox::Configuration> &session_config,
       const std::string &config, std::string &uri,
       modelbox::DestroyUriFunc &destroy_uri_func) override;
   modelbox::Status GetStreamType(const std::string &config,
diff --git a/src/drivers/devices/cpu/flowunit/data_source_parser/parser_plugin/vcn_restful_source_parser/vcn_restful_source_parser.cc b/src/drivers/devices/cpu/flowunit/data_source_parser/parser_plugin/vcn_restful_source_parser/vcn_restful_source_parser.cc
@@ -29,8 +29,8 @@ modelbox::Status VcnRestfulSourceParser::Init(
   keep_alive_interval_ =
       opts->GetInt32("vcn_keep_alive_interval_sec", keep_alive_interval_);
 
-  MBLOG_INFO << "vcn restful source parser config retry_enabled:" << retry_enabled_
-             << " retry_interval:" << retry_interval_
+  MBLOG_INFO << "vcn restful source parser config retry_enabled:"
+             << retry_enabled_ << " retry_interval:" << retry_interval_
              << " retry_max_times:" << retry_max_times_
              << " keep_alive_interval:" << keep_alive_interval_;
 
@@ -43,6 +43,7 @@ modelbox::Status VcnRestfulSourceParser::Deinit() {
 
 modelbox::Status VcnRestfulSourceParser::Parse(
     const std::shared_ptr<modelbox::SessionContext> &session_context,
+    const std::shared_ptr<modelbox::Configuration> &session_config,
     const std::string &config, std::string &uri,
     modelbox::DestroyUriFunc &destroy_uri_func) {
   modelbox::VcnRestfulInfo vcn_info;
diff --git a/src/drivers/devices/cpu/flowunit/data_source_parser/parser_plugin/vcn_restful_source_parser/vcn_restful_source_parser.h b/src/drivers/devices/cpu/flowunit/data_source_parser/parser_plugin/vcn_restful_source_parser/vcn_restful_source_parser.h
@@ -40,6 +40,7 @@ class VcnRestfulSourceParser : public modelbox::DataSourceParserPlugin {
 
   modelbox::Status Parse(
       const std::shared_ptr<modelbox::SessionContext> &session_context,
+      const std::shared_ptr<modelbox::Configuration> &session_config,
       const std::string &config, std::string &uri,
       modelbox::DestroyUriFunc &destroy_uri_func) override;
 
diff --git a/src/drivers/devices/cpu/flowunit/data_source_parser/parser_plugin/vcn_source_parser/vcn_source_parser.cc b/src/drivers/devices/cpu/flowunit/data_source_parser/parser_plugin/vcn_source_parser/vcn_source_parser.cc
@@ -58,6 +58,7 @@ modelbox::Status VcnSourceParser::Deinit() { return modelbox::STATUS_OK; }
 
 modelbox::Status VcnSourceParser::Parse(
     const std::shared_ptr<modelbox::SessionContext> &session_context,
+    const std::shared_ptr<modelbox::Configuration> &session_config,
     const std::string &config, std::string &uri,
     modelbox::DestroyUriFunc &destroy_uri_func) {
   modelbox::VcnInfo vcn_info;
diff --git a/src/drivers/devices/cpu/flowunit/data_source_parser/parser_plugin/vcn_source_parser/vcn_source_parser.h b/src/drivers/devices/cpu/flowunit/data_source_parser/parser_plugin/vcn_source_parser/vcn_source_parser.h
@@ -39,6 +39,7 @@ class VcnSourceParser : public modelbox::DataSourceParserPlugin {
 
   modelbox::Status Parse(
       const std::shared_ptr<modelbox::SessionContext> &session_context,
+      const std::shared_ptr<modelbox::Configuration> &session_config,
       const std::string &config, std::string &uri,
       modelbox::DestroyUriFunc &destroy_uri_func) override;
 
diff --git a/src/drivers/devices/cpu/flowunit/data_source_parser/parser_plugin/vis_source_parser/vis_source_parser.cc b/src/drivers/devices/cpu/flowunit/data_source_parser/parser_plugin/vis_source_parser/vis_source_parser.cc
@@ -63,6 +63,7 @@ modelbox::Status VisSourceParser::Deinit() { return modelbox::STATUS_OK; }
 
 modelbox::Status VisSourceParser::Parse(
     const std::shared_ptr<modelbox::SessionContext> &session_context,
+    const std::shared_ptr<modelbox::Configuration> &session_config,
     const std::string &config, std::string &uri,
     modelbox::DestroyUriFunc &destroy_uri_func) {
   VisInputInfo input_info;
diff --git a/src/drivers/devices/cpu/flowunit/data_source_parser/parser_plugin/vis_source_parser/vis_source_parser.h b/src/drivers/devices/cpu/flowunit/data_source_parser/parser_plugin/vis_source_parser/vis_source_parser.h
@@ -50,6 +50,7 @@ class VisSourceParser : public modelbox::DataSourceParserPlugin {
 
   modelbox::Status Parse(
       const std::shared_ptr<modelbox::SessionContext> &session_context,
+      const std::shared_ptr<modelbox::Configuration> &session_config,
       const std::string &config, std::string &uri,
       modelbox::DestroyUriFunc &destroy_uri_func) override;
 
diff --git a/src/libmodelbox/include/modelbox/data_source_parser_plugin.h b/src/libmodelbox/include/modelbox/data_source_parser_plugin.h
@@ -44,6 +44,7 @@ class DataSourceParserPlugin : public Driver {
   virtual Status Deinit() = 0;
 
   virtual Status Parse(const std::shared_ptr<SessionContext> &session_context,
+                       const std::shared_ptr<modelbox::Configuration> &session_config, 
                        const std::string &config, std::string &uri,
                        DestroyUriFunc &destroy_uri_func) = 0;
 
