explicitly subscribe dummy sink to TFN&TFF

Author: aalkin

Framework/Core/src/ArrowSupport.cxx

@@ -687,7 +687,7 @@ o2::framework::ServiceSpec ArrowSupport::arrowBackendSpec()
           });
           if (tfnsource == workflow.end()) { // add normal reader algorithm only if no on-the-fly generator is injected
             reader->algorithm = CommonDataProcessors::wrapWithTimesliceConsumption(PluginManager::loadAlgorithmFromPlugin("O2FrameworkAnalysisSupport", "ROOTFileReader", ctx));
-          } // otherwise the algorithm was set in injectServiceDevices
+          } // otherwise the algorithm was already set in injectServiceDevices
         }
       }
 

Framework/Core/src/WorkflowHelpers.cxx

@@ -409,13 +409,13 @@ void WorkflowHelpers::injectServiceDevices(WorkflowSpec& workflow, ConfigContext
     extraSpecs.push_back(indexBuilder);
   }
 
+  auto tfnsource = std::ranges::find_if(workflow, [](DataProcessorSpec const& spec) {
+    return std::ranges::any_of(spec.outputs, [](OutputSpec const& output) {
+      return DataSpecUtils::match(output, "TFN", "TFNumber", 0);
+    });
+  });
   // add the reader
   if (aodReader.outputs.empty() == false) {
-    auto tfnsource = std::ranges::find_if(workflow, [](DataProcessorSpec const& spec) {
-      return std::ranges::any_of(spec.outputs, [](OutputSpec const& output) {
-        return DataSpecUtils::match(output, "TFN", "TFNumber", 0);
-      });
-    });
     if (tfnsource == workflow.end()) {
       // add normal reader
       aodReader.outputs.emplace_back(OutputSpec{"TFN", "TFNumber"});
@@ -557,6 +557,12 @@ void WorkflowHelpers::injectServiceDevices(WorkflowSpec& workflow, ConfigContext
 
   std::vector<InputSpec> unmatched;
   auto forwardingDestination = ctx.options().get<std::string>("forwarding-destination");
+  // update tfnsource iterator (could be aod-reader)
+  tfnsource = std::ranges::find_if(workflow, [](DataProcessorSpec const& spec) {
+    return std::ranges::any_of(spec.outputs, [](OutputSpec const& output) {
+      return DataSpecUtils::match(output, "TFN", "TFNumber", 0);
+    });
+  });
   if (redirectedOutputsInputs.size() > 0 && forwardingDestination == "file") {
     auto fileSink = CommonDataProcessors::getGlobalFileSink(redirectedOutputsInputs, unmatched);
     if (unmatched.size() != redirectedOutputsInputs.size()) {
@@ -568,7 +574,7 @@ void WorkflowHelpers::injectServiceDevices(WorkflowSpec& workflow, ConfigContext
   } else if (forwardingDestination != "drop") {
     throw runtime_error_f("Unknown forwarding destination %s", forwardingDestination.c_str());
   }
-  if (unmatched.size() > 0 || redirectedOutputsInputs.size() > 0) {
+  if ((unmatched.size() > 0) || (redirectedOutputsInputs.size() > 0) || (tfnsource != workflow.end())) {
     std::vector<InputSpec> ignored = unmatched;
     ignored.insert(ignored.end(), redirectedOutputsInputs.begin(), redirectedOutputsInputs.end());
     for (auto& ignoredInput : ignored) {
@@ -577,8 +583,11 @@ void WorkflowHelpers::injectServiceDevices(WorkflowSpec& workflow, ConfigContext
 
     // Use the new dummy sink when the AOD reader is there
     O2_SIGNPOST_ID_GENERATE(sid, workflow_helpers);
-    if (aodReader.outputs.empty() == false) {
+    if (tfnsource != workflow.end()) {
       O2_SIGNPOST_EVENT_EMIT(workflow_helpers, sid, "injectServiceDevices", "Injecting scheduled dummy sink");
+      // if there is a tfnsource, make sure the sink has
+      ignored.emplace_back("tfn", "TFN", "TFNumber");
+      ignored.emplace_back("tff", "TFF", "TFFilename");
       extraSpecs.push_back(CommonDataProcessors::getScheduledDummySink(ignored));
     } else {
       O2_SIGNPOST_EVENT_EMIT(workflow_helpers, sid, "injectServiceDevices", "Injecting rate limited dummy sink");