AliceO2Group
diff --git a/‎Framework/Core/include/Framework/CompletionPolicyHelpers.h‎
Lines changed: 9 additions & 0 deletions b/‎Framework/Core/include/Framework/CompletionPolicyHelpers.h‎
Lines changed: 9 additions & 0 deletions
diff --git a/‎Framework/Core/src/CompletionPolicyHelpers.cxx‎
Lines changed: 23 additions & 2 deletions b/‎Framework/Core/src/CompletionPolicyHelpers.cxx‎
Lines changed: 23 additions & 2 deletions
@@ -51,6 +51,15 @@ struct CompletionPolicyHelpers {
   }
   static CompletionPolicy consumeWhenAny(std::string matchName);
 
+  /// When any of the parts of the record have been received, consume them.
+  static CompletionPolicy consumeWhenAnyWithAllConditions(const char* name, CompletionPolicy::Matcher matcher);
+  /// Default matcher applies for all devices
+  static CompletionPolicy consumeWhenAnyWithAllConditions(CompletionPolicy::Matcher matcher = [](auto const&) -> bool { return true; })
+  {
+    return consumeWhenAnyWithAllConditions("consume-any-all-conditions", matcher);
+  }
+  static CompletionPolicy consumeWhenAnyWithAllConditions(std::string matchName);
+
   /// When any of the parts of the record have been received, process the existing and free the associated payloads.
   /// This allows freeing things as early as possible, while still being able to wait
   /// all the parts before disposing the timeslice completely
 
@@ -198,6 +198,27 @@ CompletionPolicy CompletionPolicyHelpers::consumeExistingWhenAny(const char* nam
 }
 
 CompletionPolicy CompletionPolicyHelpers::consumeWhenAny(const char* name, CompletionPolicy::Matcher matcher)
+{
+  auto callback = [](InputSpan const& inputs) -> CompletionPolicy::CompletionOp {
+    for (auto& input : inputs) {
+      if (input.header != nullptr) {
+        return CompletionPolicy::CompletionOp::Consume;
+      }
+    }
+    return CompletionPolicy::CompletionOp::Wait;
+  };
+  return CompletionPolicy{name, matcher, callback, false};
+}
+
+CompletionPolicy CompletionPolicyHelpers::consumeWhenAny(std::string matchName)
+{
+  auto matcher = [matchName](DeviceSpec const& device) -> bool {
+    return std::regex_match(device.name.begin(), device.name.end(), std::regex(matchName));
+  };
+  return consumeWhenAny(matcher);
+}
+
+CompletionPolicy CompletionPolicyHelpers::consumeWhenAnyWithAllConditions(const char* name, CompletionPolicy::Matcher matcher)
 {
   auto callback = [](InputSpan const& inputs, std::vector<InputSpec> const& specs) -> CompletionPolicy::CompletionOp {
     bool canConsume = false;
@@ -258,12 +279,12 @@ CompletionPolicy CompletionPolicyHelpers::consumeWhenAny(const char* name, Compl
   return CompletionPolicy{name, matcher, callback, false};
 }
 
-CompletionPolicy CompletionPolicyHelpers::consumeWhenAny(std::string matchName)
+CompletionPolicy CompletionPolicyHelpers::consumeWhenAnyWithAllConditions(std::string matchName)
 {
   auto matcher = [matchName](DeviceSpec const& device) -> bool {
     return std::regex_match(device.name.begin(), device.name.end(), std::regex(matchName));
   };
-  return consumeWhenAny(matcher);
+  return consumeWhenAnyWithAllConditions(matcher);
 }
 
 CompletionPolicy CompletionPolicyHelpers::processWhenAny(const char* name, CompletionPolicy::Matcher matcher)