Fix forwarding inputs to consumers which are time-pipelined

Author: knopers8

Framework/Core/include/Framework/ForwardRoute.h

@@ -23,6 +23,8 @@ namespace framework
 /// the InputSpec @a matcher matches an input which should also go to
 /// @a channel
 struct ForwardRoute {
+  size_t timeslice;
+  size_t maxTimeslices;
   InputSpec matcher;
   std::string channel;
 };

Framework/Core/src/DataProcessingDevice.cxx

@@ -389,9 +389,8 @@ bool DataProcessingDevice::tryDispatchComputation()
         LOG(DEBUG) << dh->dataOrigin.str;
         LOG(DEBUG) << dh->dataDescription.str;
         LOG(DEBUG) << dh->subSpecification;
-        if (DataSpecUtils::match(forward.matcher, dh->dataOrigin,
-                                 dh->dataDescription,
-                                 dh->subSpecification)) {
+        if (DataSpecUtils::match(forward.matcher, dh->dataOrigin, dh->dataDescription, dh->subSpecification)
+            && (dph->startTime % forward.maxTimeslices) == forward.timeslice) {
 
           if (header.get() == nullptr) {
             LOG(ERROR) << "Missing header!";

Framework/Core/src/DeviceSpecHelpers.cxx

@@ -255,6 +255,8 @@ void DeviceSpecHelpers::processOutEdgeActions(std::vector<DeviceSpec>& devices,
       ForwardRoute route;
       route.matcher = workflow[edge.consumer].inputs[edge.consumerInputIndex];
       route.channel = channel.name;
+      route.timeslice = edge.timeIndex;
+      route.maxTimeslices = consumer.maxInputTimeslices;
       device.forwards.emplace_back(route);
     }
   };