Adding Pipes

Src/Data/Myc.Data.Scalar.pas

@@ -200,7 +200,7 @@ type
         function GetTotalCount: Int64;
         {$endregion}
         procedure Add(const Item: IKeywordMapping<TScalar>; Lookback: Int64 = -1); overload;
-        procedure Add(const Items: TArray<TScalar.TValue>; Lookback: Int64 = -1); overload;
+        procedure Add(const Items: array of TScalar.TValue; Lookback: Int64 = -1); overload;
         property RecordCount: Int64 read GetRecordCount;
         property TotalCount: Int64 read GetTotalCount;
     end;
@@ -235,13 +235,12 @@ type
         function GetDef: IScalarRecordDefinition; inline;
         function GetTotalCount: Int64; inline;
         function GetItemRef(Idx: Integer): TChunkArray<TScalar.TValue>.PT; inline;
-        function GetMapping: IKeywordMapping<ISeries>;
     public
         constructor Create(const ADef: IScalarRecordDefinition);
         destructor Destroy; override;
 
         procedure Add(const Item: IKeywordMapping<TScalar>; Lookback: Int64 = -1); overload;
-        procedure Add(const Items: TArray<TScalar.TValue>; Lookback: Int64 = -1); overload;
+        procedure Add(const Items: array of TScalar.TValue; Lookback: Int64 = -1); overload;
 
         property RecordCount: Int64 read GetRecordCount;
         property Def: IScalarRecordDefinition read GetDef;
@@ -829,7 +828,7 @@ begin
     inc(FTotalCount);
 end;
 
-procedure TScalarRecordSeries.Add(const Items: TArray<TScalar.TValue>; Lookback: Int64 = -1);
+procedure TScalarRecordSeries.Add(const Items: array of TScalar.TValue; Lookback: Int64 = -1);
 begin
     Assert(Length(Items) = FDef.Count, 'Array does not fit series definition');
 
@@ -864,11 +863,6 @@ begin
     Result := FArray.ItemRef[(FArray.Count - len) - (Idx * len)];
 end;
 
-function TScalarRecordSeries.GetMapping: IKeywordMapping<ISeries>;
-begin
-    Result := Self;
-end;
-
 function TScalarRecordSeries.GetTotalCount: Int64;
 begin
     Result := FTotalCount;

Src/Data/Myc.Data.Stream.Pipes.pas

@@ -77,11 +77,12 @@ type
     // Acts as IStream (Producer) for downstream consumers and manages internal inputs.
     TPipeStream = class(TInterfacedObject, IStream)
         type
-            TProc = reference to procedure(const Series: TArray<ISeries>);
+            TEmitProc = reference to procedure(const Data: array of TScalar.TValue);
+            TProc = reference to procedure(const Sources: TArray<ISeries>; const Emit: TEmitProc);
     private
-        FConfig: TPipeConfig;
         FSources: TArray<TPipeSource>;
         FObservers: TMycNotifyList<IStreamObserver>;
+        FSourceSeries: TArray<ISeries>;
         FSeries: IWriteableScalarRecordSeries;
 
         // Barrier State
@@ -90,17 +91,21 @@ type
         // Execution Logic
         // The compiled lambda function representing (fn [inputs...] ...)
         FLambda: TProc;
-        FLambdaArgs: TArray<ISeries>;
 
         procedure CheckBarrierAndFire(CurrentCycle: Int64);
         function GetSeries: IScalarRecordSeries;
 
     public
-        constructor Create(const AConfig: TPipeConfig; const ASources: TArray<IStream>; const ALambda: TProc);
+        constructor Create(
+            const AConfig: TPipeConfig;
+            const ADef: IScalarRecordDefinition;
+            const ASources: TArray<IStream>;
+            const ALambda: TProc
+        );
         destructor Destroy; override;
 
         // Injected into lambda scope as "emit"
-        procedure Emit(const Value: IScalarRecord);
+        procedure Emit(const Data: array of TScalar.TValue);
 
         // IStream Implementation
         function Subscribe(const Observer: IStreamObserver): TSubscriptionTag;
@@ -138,20 +143,25 @@ end;
 
 { TPipeStream }
 
-constructor TPipeStream.Create(const AConfig: TPipeConfig; const ASources: TArray<IStream>; const ALambda: TProc);
+constructor TPipeStream.Create(
-var
+    const AConfig: TPipeConfig;
-    fields: TArray<TScalarRecordField>;
+    const ADef: IScalarRecordDefinition;
+    const ASources: TArray<IStream>;
+    const ALambda: TProc
+);
 begin
     inherited Create;
-    FConfig := AConfig;
     FLambda := ALambda;
     FLastFiredCycleID := -1;
 
     Assert(Length(AConfig) = Length(ASources));
 
-    SetLength(FSources, Length(ASources));
+    FSeries := TScalarRecordSeries.Create(ADef);
 
     // Construct record series definition from source config
+    SetLength(FSources, Length(ASources));
+    SetLength(FSourceSeries, Length(ASources));
+
     var n := 0;
     for var i := 0 to High(AConfig) do
     begin
@@ -160,22 +170,14 @@ begin
         inc(n, Length(AConfig[i]));
     end;
 
-    SetLength(FLambdaArgs, n);
+    SetLength(FSourceSeries, n);
-
     n := 0;
     for var i := 0 to High(AConfig) do
-    begin
         for var j := 0 to High(AConfig[i]) do
         begin
-            fields[n] := AConfig[i][j];
+            FSourceSeries[n] := ASources[i].Series.Fields[AConfig[i][j].Key];
-
-            FLambdaArgs[n] := ASources[i].Series[j].Value;
-
             inc(n);
         end;
-    end;
-
-    FSeries := TScalarRecordSeries.Create(TScalarRecord.TRegistry.Intern(fields));
 end;
 
 destructor TPipeStream.Destroy;
@@ -210,52 +212,58 @@ begin
 end;
 
 procedure TPipeStream.CheckBarrierAndFire(CurrentCycle: Int64);
-begin
-    // Barrier Check
-    for var src in FSources do
-    begin
-        if src.LastSeenCycle < CurrentCycle then
-            exit;
-    end;
-
-    // --- Barrier Open ---
-
-    if FLastFiredCycleID >= CurrentCycle then
-        exit;
-
-    FLastFiredCycleID := CurrentCycle;
-
-    // Execute Lambda
-    if Assigned(FLambda) then
-    begin
-        try
-            FLambda(FLambdaArgs);
-        except
-            // Error handling strategy: Propagate/Crash for now.
-            raise;
-        end;
-    end;
-end;
-
-procedure TPipeStream.Emit(const Value: IScalarRecord);
 begin
     FObservers.Lock;
     try
-        FSeries.Add(Value);
+        // Barrier Check
-        var signal := TStreamSignal.Create(skData, FLastFiredCycleID);
+        for var src in FSources do
+        begin
+            if src.LastSeenCycle < CurrentCycle then
+                exit;
+        end;
 
-        FObservers.Notify(
+        // --- Barrier Open ---
-            function(const Obs: IStreamObserver): Boolean
+
-            begin
+        if FLastFiredCycleID >= CurrentCycle then
-                Obs.OnSignal(signal);
+            exit;
-                Result := True;
+
-            end
+        FLastFiredCycleID := CurrentCycle;
-        );
+
+        // Execute Lambda
+        if Assigned(FLambda) then
+        begin
+            try
+                FLambda(FSourceSeries, Emit);
+            except
+                // Error handling strategy: Propagate/Crash for now.
+                raise;
+            end;
+        end;
     finally
         FObservers.Release;
     end;
 end;
 
+procedure TPipeStream.Emit(const Data: array of TScalar.TValue);
+begin
+    Assert(Length(Data) = FSeries.Def.Count);
+
+    // Emit has to be called from the Lambda, which is called in CheckBarrierAndFire!
+    Assert(FObservers.IsLocked);
+
+    FSeries.Add(Data);
+
+    var signal := TStreamSignal.Create(skData, FLastFiredCycleID);
+
+    FObservers.Notify(
+        function(const Obs: IStreamObserver): Boolean
+        begin
+            Obs.OnSignal(signal);
+            Result := True;
+        end
+    );
+end;
+
 function TPipeStream.GetSeries: IScalarRecordSeries;
 begin
     Result := FSeries;