diff --git a/src/specificClass.js b/src/specificClass.js
index 3ad3fa5..aba5570 100644
--- a/src/specificClass.js
+++ b/src/specificClass.js
@@ -68,6 +68,10 @@ class Machine {
       this.updatePosition();
     });
 
+    // used for holding the source and sink unit operations or other object with setInfluent / getEffluent method for e.g. recirculation.
+    this.upstreamReactor = null;
+    this.downstreamReactor = null;
+
     this.child = {}; // object to hold child information so we know on what to subscribe
     this.childRegistrationUtils = new childRegistrationUtils(this); // Child registration utility
 
@@ -92,7 +96,7 @@ class Machine {
 
   _connectMeasurement(measurementChild) {
     if (!measurementChild) {
-      this.logger.warn("Invalid measurement provided.");
+      this.logger.error("Invalid measurement provided.");
       return;
     }
 
@@ -107,14 +111,12 @@ class Machine {
     measurementChild.measurements.emitter.on(eventName, (eventData) => {
       this.logger.debug(`🔄 ${position} ${measurementType} from ${eventData.childName}: ${eventData.value} ${eventData.unit}`);
 
-
-      console.log(` Emitting... ${eventName} with data:`);
       // Store directly in parent's measurement container
       this.measurements
         .type(measurementType)
         .variant("measured")
         .position(position)
-        .value(eventData.value, eventData.timestamp, eventData.unit);
+        .value(eventData.value, eventData.timestamp, eventData.unit);      
       
       // Call the appropriate handler
       switch (measurementType) {
@@ -135,26 +137,31 @@ class Machine {
   }
 
   _connectReactor(reactorChild) {
-    this.logger.error("Reactor child not implemented yet.");
+    if (!reactorChild) {
+      this.logger.error("Invalid measurement provided.");
+      return;
+    }
+
+    this.downstreamReactor = reactorChild; // downstream from the pumps perpective
   }
 
-//---------------- END child stuff -------------//
+  //---------------- END child stuff -------------//
 
-    // Method to assess drift using errorMetrics
-    assessDrift(measurement, processMin, processMax) {
-      this.logger.debug(`Assessing drift for measurement: ${measurement} processMin: ${processMin} processMax: ${processMax}`);
-      const predictedMeasurement = this.measurements.type(measurement).variant("predicted").position("downstream").getAllValues().values;
-      const measuredMeasurement = this.measurements.type(measurement).variant("measured").position("downstream").getAllValues().values;
+  // Method to assess drift using errorMetrics
+  assessDrift(measurement, processMin, processMax) {
+    this.logger.debug(`Assessing drift for measurement: ${measurement} processMin: ${processMin} processMax: ${processMax}`);
+    const predictedMeasurement = this.measurements.type(measurement).variant("predicted").position("downstream").getAllValues().values;
+    const measuredMeasurement = this.measurements.type(measurement).variant("measured").position("downstream").getAllValues().values;
 
-      if (!predictedMeasurement || !measuredMeasurement) return null;
-      
-      return this.errorMetrics.assessDrift(
-        predictedMeasurement,
-        measuredMeasurement,
-        processMin,
-        processMax
-      );
-    }
+    if (!predictedMeasurement || !measuredMeasurement) return null;
+    
+    return this.errorMetrics.assessDrift(
+      predictedMeasurement,
+      measuredMeasurement,
+      processMin,
+      processMax
+    );
+  }
 
   reverseCurve(curve) {
     const reversedCurve = {};
@@ -499,12 +506,17 @@ class Machine {
 
   // NEW: Flow handler
   updateMeasuredFlow(value, position, context = {}) {
+
     if (!this._isOperationalState()) {
       this.logger.warn(`Machine not operational, skipping flow update from ${context.childName || 'unknown'}`);
       return;
     }
 
     this.logger.debug(`Flow update: ${value} at ${position} from ${context.childName || 'child'}`);
+
+    if (this.upstreamReactor && this.downstreamReactor){
+      this._updateConnectedReactor();
+    }
     
     // Store in parent's measurement container
     this.measurements.type("flow").variant("measured").position(position).value(value, context.timestamp, context.unit);
@@ -515,6 +527,10 @@ class Machine {
     }
   }
 
+  _updateConnectedReactor() {
+    this.downstreamReactor.setInfluent = this.upstreamReactor.getEffluent[1];
+  }
+
   // Helper method for operational state check
   _isOperationalState() {
     const state = this.state.getCurrentState();