Automatic InputSpecs generation for a policy

Author: knopers8

Framework/Core/include/Framework/DataSampling.h

@@ -78,6 +78,10 @@ class DataSampling
   static void CustomizeInfrastructure(std::vector<CompletionPolicy>&);
   /// \brief Applies blocking/nonblocking data sampling configuration to the workflow.
   static void CustomizeInfrastructure(std::vector<ChannelConfigurationPolicy>&);
+  /// \brief Provides InputSpecs to receive data for given DataSamplingPolicy
+  static std::vector<InputSpec> InputSpecsForPolicy(const std::string& policiesSource, const std::string& policyName);
+  /// \brief Provides InputSpecs to receive data for given DataSamplingPolicy
+  static std::vector<InputSpec> InputSpecsForPolicy(configuration::ConfigurationInterface* const config, const std::string& policyName);
 
  private:
   // Internal functions, used by GenerateInfrastructure()

Framework/Core/src/DataSampling.cxx

@@ -112,5 +112,38 @@ void DataSampling::CustomizeInfrastructure(std::vector<ChannelConfigurationPolic
   // now it cannot be done, since matching is possible only using data processors names
 }
 
+std::vector<InputSpec> DataSampling::InputSpecsForPolicy(const std::string& policiesSource, const std::string& policyName)
+{
+  std::unique_ptr<ConfigurationInterface> config = ConfigurationFactory::getConfiguration(policiesSource);
+  return InputSpecsForPolicy(config.get(), policyName);
+}
+
+std::vector<InputSpec> DataSampling::InputSpecsForPolicy(ConfigurationInterface* const config, const std::string& policyName)
+{
+  std::vector<InputSpec> inputs;
+  auto policiesTree = config->getRecursive("dataSamplingPolicies");
+
+  for (auto&& policyConfig : policiesTree) {
+    if (policyConfig.second.get<std::string>("id") == policyName) {
+      DataSamplingPolicy policy(policyConfig.second);
+      if (policy.getSubSpec() == -1) {
+        //fixme: support it, when wildcards are available
+        LOG(WARNING) << "InputSpecsForPolicy does not support subscriptions to all subSpecs yet.";
+      }
+      for (const auto& path : policy.getPathMap()) {
+        inputs.push_back(
+          InputSpec{
+            path.second.binding.value,
+            path.second.origin,
+            path.second.description,
+            path.second.subSpec,
+            path.second.lifetime });
+      }
+    }
+    break;
+  }
+  return inputs;
+}
+
 } // namespace framework
 } // namespace o2

Framework/Core/test/test_DataSampling.cxx

@@ -235,3 +235,19 @@ BOOST_AUTO_TEST_CASE(DataSamplingFairMq)
   BOOST_REQUIRE(channelConfig != disp->options.end());
 }
 
+BOOST_AUTO_TEST_CASE(InputSpecsForPolicy)
+{
+  std::string configFilePath = "json://" + std::string(getenv("O2_ROOT")) + "/share/tests/test_DataSampling.json";
+  std::vector<InputSpec> inputs = DataSampling::InputSpecsForPolicy(configFilePath, "tpcclusters");
+
+  BOOST_CHECK_EQUAL(inputs.size(), 2);
+  BOOST_CHECK_EQUAL(inputs[0], (InputSpec{ "clusters_p", "DS", "tpcclusters-1", static_cast<DataHeader::SubSpecificationType>(-1) }));
+  BOOST_CHECK_EQUAL(inputs[0].binding, "clusters_p");
+  BOOST_CHECK_EQUAL(inputs[1], (InputSpec{ "clusters", "DS", "tpcclusters-0", static_cast<DataHeader::SubSpecificationType>(-1) }));
+  BOOST_CHECK_EQUAL(inputs[1].binding, "clusters");
+
+  std::unique_ptr<ConfigurationInterface> config = ConfigurationFactory::getConfiguration(configFilePath);
+  inputs = DataSampling::InputSpecsForPolicy(config.get(), "tpcclusters");
+
+  BOOST_CHECK_EQUAL(inputs.size(), 2);
+}