Adding Pipes

Src/AST/Myc.Ast.Evaluator.pas

@@ -431,9 +431,6 @@ begin
         vkRecordSeries:
         begin
             recSeries := baseValue.AsRecordSeries;
-            if (index < 0) or (index >= recSeries.TotalCount) then
-                raise EEvaluatorException
-                    .CreateFmt('Index %d is out of bounds for series with %d elements.', [index, recSeries.TotalCount]);
 
             fieldCount := recSeries.Def.Count;
             SetLength(values, fieldCount);
@@ -441,7 +438,12 @@ begin
             for i := 0 to fieldCount - 1 do
             begin
                 key := recSeries.Def.Items[i].Key;
-                memberSeries := recSeries[key];
+                memberSeries := recSeries.Fields[key];
+
+                if (index < 0) or (index >= memberSeries.TotalCount) then
+                    raise EEvaluatorException
+                        .CreateFmt('Index %d is out of bounds for series with %d elements.', [index, memberSeries.TotalCount]);
+
                 scalarValue := memberSeries[index];
                 values[i] := scalarValue.Value;
             end;
@@ -460,7 +462,7 @@ begin
     baseValue := Node.Base.Accept(Self);
 
     case baseValue.Kind of
-        vkRecordSeries: Result := TDataValue.FromSeries(baseValue.AsRecordSeries[Node.Member.Value]);
+        vkRecordSeries: Result := TDataValue.FromSeries(baseValue.AsRecordSeries.Fields[Node.Member.Value]);
         vkScalarRecord: Result := TDataValue(baseValue.AsScalarRecord.Fields[Node.Member.Value]);
         vkGenericRecord: Result := TDataValue(baseValue.AsGenericRecord.Fields[Node.Member.Value]);
     else
@@ -583,7 +585,7 @@ begin
 
     case seriesValue.Kind of
         vkSeries: len := seriesValue.AsSeries.Count;
-        vkRecordSeries: len := seriesValue.AsRecordSeries.Count;
+        vkRecordSeries: len := seriesValue.AsRecordSeries.RecordCount;
     else
         raise EEvaluatorException
             .CreateFmt('Cannot get length of type %s.', [GetEnumName(TypeInfo(TDataValueKind), Ord(seriesValue.Kind))]);

Src/AST/Myc.Ast.Nodes.pas

@@ -141,7 +141,11 @@ type
         akAddSeriesItem,
         akSeriesLength,
         akRecur,
-        akNop
+        akNop,
+        // Pipes
+        akPipe,
+        akPipeInput,
+        akPipeInputList
     );
 
     TAstNodeKindHelper = record helper for TAstNodeKind
@@ -452,10 +456,38 @@ type
         property Series: IIdentifierNode read GetSeries;
     end;
 
+    // Represents a single input definition within a pipe: "btc [:Close]"
+    // - StreamSource: The identifier of the stream (e.g., "btc")
+    // - Selector: Optional keyword path to select a field (e.g., :Close)
+    IPipeInputNode = interface(IAstTypedNode)
+        {$region 'private'}
+        function GetStreamSource: IIdentifierNode;
+        function GetSelector: IKeywordNode; // Can be nil if the whole record is used
+        {$endregion}
+        property StreamSource: IIdentifierNode read GetStreamSource;
+        property Selector: IKeywordNode read GetSelector;
+    end;
+
+    // A list of pipe inputs
+    IPipeInputList = interface(INodeList<IPipeInputNode>)
+    end;
+
+    // The main pipe definition: (pipe [inputs...] (fn ...))
+    IPipeNode = interface(IAstTypedNode)
+        {$region 'private'}
+        function GetInputs: IPipeInputList;
+        function GetTransformation: ILambdaExpressionNode;
+        {$endregion}
+        property Inputs: IPipeInputList read GetInputs;
+        property Transformation: ILambdaExpressionNode read GetTransformation;
+    end;
+
     IParameterList = interface(INodeList<IIdentifierNode>)
     end;
+
     IArgumentList = interface(INodeList<IAstNode>)
     end;
+
     IExpressionList = interface(INodeList<IAstNode>)
     end;
 
@@ -491,6 +523,12 @@ type
         function VisitSeriesLength(const Node: ISeriesLengthNode): TDataValue;
         function VisitRecurNode(const Node: IRecurNode): TDataValue;
         function VisitNop(const Node: INopNode): TDataValue;
+
+    // Pipes
+    // Not implemented yet. that's ok, it just needs to compile
+    //        function VisitPipeInput(const Node: IPipeInputNode): TDataValue;
+    //        function VisitPipeInputList(const Node: IPipeInputList): TDataValue;
+    //        function VisitPipe(const Node: IPipeNode): TDataValue;
     end;
 
     IEvaluatorVisitor = interface(IAstVisitor)
@@ -502,9 +540,10 @@ type
     TAstNode = class(TInterfacedObject, IAstNode)
     private
         FIdentity: IAstIdentity;
+        function GetIdentity: IAstIdentity;
+    protected
         function GetKind: TAstNodeKind; virtual; abstract;
         function GetIsTyped: Boolean; virtual;
-        function GetIdentity: IAstIdentity;
     public
         constructor Create(const AIdentity: IAstIdentity);
         function Accept(const Visitor: IAstVisitor): TDataValue; virtual; abstract;
@@ -551,6 +590,7 @@ type
     private
         FStaticType: IStaticType;
         function GetStaticType: IStaticType;
+    protected
         function GetIsTyped: Boolean; override;
     public
         constructor Create(const AStaticType: IStaticType; const AIdentity: IAstIdentity);
@@ -617,6 +657,7 @@ type
         FValue: IAstNode;
         function GetKey: IKeywordNode;
         function GetValue: IAstNode;
+    protected
         function GetKind: TAstNodeKind; override;
     public
         constructor Create(const AKey: IKeywordNode; const AValue: IAstNode; const AIdentity: IAstIdentity);
@@ -638,6 +679,7 @@ type
     private
         FConstIdentity: IConstantIdentity; // Typed reference for fast access
         function GetValue: TDataValue;
+    protected
         function GetKind: TAstNodeKind; override;
     public
         constructor Create(const AIdentity: IConstantIdentity; const AStaticType: IStaticType);
@@ -651,6 +693,7 @@ type
         FAddress: TResolvedAddress;
         function GetAddress: TResolvedAddress;
         function GetName: string;
+    protected
         function GetKind: TAstNodeKind; override;
     public
         constructor Create(const AIdentity: INamedIdentity; const AAddress: TResolvedAddress; const AStaticType: IStaticType);
@@ -662,6 +705,7 @@ type
     private
         FKeywordIdentity: IKeywordIdentity;
         function GetValue: IKeyword;
+    protected
         function GetKind: TAstNodeKind; override;
     public
         constructor Create(const AIdentity: IKeywordIdentity);
@@ -673,6 +717,7 @@ type
     private
         FDefIdentity: IDefinitionIdentity;
         function GetDefinition: String;
+    protected
         function GetKind: TAstNodeKind; override;
     public
         constructor Create(const AIdentity: IDefinitionIdentity; const AStaticType: IStaticType);
@@ -688,6 +733,7 @@ type
         function GetCondition: IAstNode;
         function GetThenBranch: IAstNode;
         function GetElseBranch: IAstNode;
+    protected
         function GetKind: TAstNodeKind; override;
     public
         constructor Create(
@@ -705,6 +751,7 @@ type
         FElseBranch: IAstNode;
         function GetPairs: TArray<TCondPair>;
         function GetElseBranch: IAstNode;
+    protected
         function GetKind: TAstNodeKind; override;
     public
         constructor Create(
@@ -732,6 +779,7 @@ type
         function GetUpvalues: TArray<TResolvedAddress>;
         function GetHasNestedLambdas: Boolean;
         function GetIsPure: Boolean;
+    protected
         function GetKind: TAstNodeKind; override;
     public
         constructor Create(
@@ -760,6 +808,7 @@ type
         function GetIsTailCall: Boolean;
         function GetStaticTarget: TDataValue.TFunc;
         function GetIsTargetPure: Boolean;
+    protected
         function GetKind: TAstNodeKind; override;
     public
         constructor Create(
@@ -783,6 +832,7 @@ type
         function GetName: IIdentifierNode;
         function GetParameters: IParameterList;
         function GetBody: IAstNode;
+    protected
         function GetKind: TAstNodeKind; override;
     public
         constructor Create(
@@ -801,6 +851,7 @@ type
         FExpandedBody: IAstNode;
         function GetCallNode: IFunctionCallNode;
         function GetExpandedBody: IAstNode;
+    protected
         function GetKind: TAstNodeKind; override;
     public
         constructor Create(const ACallNode: IFunctionCallNode; const AExpandedBody: IAstNode; const AIdentity: IAstIdentity);
@@ -812,6 +863,7 @@ type
     private
         FExpression: IAstNode;
         function GetExpression: IAstNode;
+    protected
         function GetKind: TAstNodeKind; override;
     public
         constructor Create(const AExpression: IAstNode; const AIdentity: IAstIdentity);
@@ -823,6 +875,7 @@ type
     private
         FExpression: IAstNode;
         function GetExpression: IAstNode;
+    protected
         function GetKind: TAstNodeKind; override;
     public
         constructor Create(const AExpression: IAstNode; const AIdentity: IAstIdentity);
@@ -834,6 +887,7 @@ type
     private
         FExpression: IQuasiquoteNode;
         function GetExpression: IQuasiquoteNode;
+    protected
         function GetKind: TAstNodeKind; override;
     public
         constructor Create(const AExpression: IQuasiquoteNode; const AIdentity: IAstIdentity);
@@ -845,6 +899,7 @@ type
     private
         FArguments: IArgumentList;
         function GetArguments: IArgumentList;
+    protected
         function GetKind: TAstNodeKind; override;
     public
         constructor Create(const AArguments: IArgumentList; const AStaticType: IStaticType; const AIdentity: IAstIdentity);
@@ -856,6 +911,7 @@ type
     private
         FExpressions: IExpressionList;
         function GetExpressions: IExpressionList;
+    protected
         function GetKind: TAstNodeKind; override;
     public
         constructor Create(const AExpressions: IExpressionList; const AStaticType: IStaticType; const AIdentity: IAstIdentity);
@@ -870,6 +926,7 @@ type
         function GetTarget: IAstNode;
         function GetInitializer: IAstNode;
         function GetIsBoxed: Boolean;
+    protected
         function GetKind: TAstNodeKind; override;
     public
         constructor Create(
@@ -888,6 +945,7 @@ type
         FTarget, FValue: IAstNode;
         function GetTarget: IAstNode;
         function GetValue: IAstNode;
+    protected
         function GetKind: TAstNodeKind; override;
     public
         constructor Create(const ATarget, AValue: IAstNode; const AStaticType: IStaticType; const AIdentity: IAstIdentity);
@@ -900,6 +958,7 @@ type
         FBase, FIndex: IAstNode;
         function GetBase: IAstNode;
         function GetIndex: IAstNode;
+    protected
         function GetKind: TAstNodeKind; override;
     public
         constructor Create(const ABase, AIndex: IAstNode; const AStaticType: IStaticType; const AIdentity: IAstIdentity);
@@ -913,6 +972,7 @@ type
         FMember: IKeywordNode;
         function GetBase: IAstNode;
         function GetMember: IKeywordNode;
+    protected
         function GetKind: TAstNodeKind; override;
     public
         constructor Create(
@@ -933,6 +993,7 @@ type
         function GetFields: IRecordFieldList;
         function GetGenericDefinition: IGenericRecordDefinition;
         function GetScalarDefinition: IScalarRecordDefinition;
+    protected
         function GetKind: TAstNodeKind; override;
     public
         constructor Create(
@@ -953,6 +1014,7 @@ type
         function GetSeries: IIdentifierNode;
         function GetValue: IAstNode;
         function GetLookback: IAstNode;
+    protected
         function GetKind: TAstNodeKind; override;
     public
         constructor Create(
@@ -969,6 +1031,7 @@ type
     private
         FSeries: IIdentifierNode;
         function GetSeries: IIdentifierNode;
+    protected
         function GetKind: TAstNodeKind; override;
     public
         constructor Create(const ASeries: IIdentifierNode; const AStaticType: IStaticType; const AIdentity: IAstIdentity);
@@ -977,7 +1040,7 @@ type
     end;
 
     TNopNode = class(TAstTypedNode, INopNode)
-    private
+    protected
         function GetKind: TAstNodeKind; override;
     public
         constructor Create(const AStaticType: IStaticType; const AIdentity: IAstIdentity);
@@ -985,6 +1048,49 @@ type
         function AsNop: INopNode; override;
     end;
 
+    TPipeInputNode = class(TAstTypedNode, IPipeInputNode)
+    private
+        FStreamSource: IIdentifierNode;
+        FSelector: IKeywordNode;
+        function GetStreamSource: IIdentifierNode;
+        function GetSelector: IKeywordNode;
+    protected
+        function GetKind: TAstNodeKind; override;
+    public
+        constructor Create(
+            const AStreamSource: IIdentifierNode;
+            const ASelector: IKeywordNode;
+            const AIdentity: IAstIdentity;
+            const AStaticType: IStaticType
+        );
+        function Accept(const Visitor: IAstVisitor): TDataValue; override;
+    end;
+
+    TPipeInputList = class(TAstNodeList<IPipeInputNode>, IPipeInputList)
+    protected
+        function GetKind: TAstNodeKind; override;
+    public
+        function Accept(const Visitor: IAstVisitor): TDataValue; override;
+    end;
+
+    TPipeNode = class(TAstTypedNode, IPipeNode)
+    private
+        FInputs: IPipeInputList;
+        FTransformation: ILambdaExpressionNode;
+        function GetInputs: IPipeInputList;
+        function GetTransformation: ILambdaExpressionNode;
+    protected
+        function GetKind: TAstNodeKind; override;
+    public
+        constructor Create(
+            const AInputs: IPipeInputList;
+            const ATransformation: ILambdaExpressionNode;
+            const AIdentity: IAstIdentity;
+            const AStaticType: IStaticType
+        );
+        function Accept(const Visitor: IAstVisitor): TDataValue; override;
+    end;
+
 implementation
 
 uses
@@ -2260,4 +2366,84 @@ begin
         Inc(FIndex);
 end;
 
+{ TPipeInputNode }
+
+constructor TPipeInputNode.Create(
+    const AStreamSource: IIdentifierNode;
+    const ASelector: IKeywordNode;
+    const AIdentity: IAstIdentity;
+    const AStaticType: IStaticType
+);
+begin
+    inherited Create(AStaticType, AIdentity);
+    FStreamSource := AStreamSource;
+    FSelector := ASelector;
+end;
+
+function TPipeInputNode.Accept(const Visitor: IAstVisitor): TDataValue;
+begin
+    //    Result := Visitor.VisitPipeInput(Self);
+end;
+
+function TPipeInputNode.GetKind: TAstNodeKind;
+begin
+    Result := akPipeInput;
+end;
+
+function TPipeInputNode.GetSelector: IKeywordNode;
+begin
+    Result := FSelector;
+end;
+
+function TPipeInputNode.GetStreamSource: IIdentifierNode;
+begin
+    Result := FStreamSource;
+end;
+
+{ TPipeInputList }
+
+function TPipeInputList.Accept(const Visitor: IAstVisitor): TDataValue;
+begin
+    //    Result := Visitor.VisitPipeInputList(Self);
+end;
+
+function TPipeInputList.GetKind: TAstNodeKind;
+begin
+    Result := akPipeInputList;
+end;
+
+{ TPipeNode }
+
+constructor TPipeNode.Create(
+    const AInputs: IPipeInputList;
+    const ATransformation: ILambdaExpressionNode;
+    const AIdentity: IAstIdentity;
+    const AStaticType: IStaticType
+);
+begin
+    inherited Create(AStaticType, AIdentity);
+    FInputs := AInputs;
+    FTransformation := ATransformation;
+end;
+
+function TPipeNode.Accept(const Visitor: IAstVisitor): TDataValue;
+begin
+    //    Result := Visitor.VisitPipe(Self);
+end;
+
+function TPipeNode.GetInputs: IPipeInputList;
+begin
+    Result := FInputs;
+end;
+
+function TPipeNode.GetKind: TAstNodeKind;
+begin
+    Result := akPipe;
+end;
+
+function TPipeNode.GetTransformation: ILambdaExpressionNode;
+begin
+    Result := FTransformation;
+end;
+
 end.

Src/AST/Myc.Ast.Pipes.pas

@@ -0,0 +1,272 @@
+(*TODO
+Ich möchte ein neues Sprachkonzept einbauen. Folgendes Skript skizziert, was möglich sein soll:
+
+; Record-Streams sind Record-Series, an die "Pipes" andocken können (Observer) um neue Daten zu bekommen
+; Es handelt sich um reaktives Producer-Consumer-Pattern
+
+;; Beispiel: SMA auf Close
+(def btc (create-ticker "BTCUSD")) ; ein Open-High-Low-Close-Record-Stream
+(def sma (create-sma 20)) ; erzeugt eine Lambda (sma(price)), die den sma20 berechnet
+
+
+; Das Ergebnis eines pipe-Befehls ist wieder ein Record-Stream.
+; btc-sma ist also ein Record-Stream, der {:ma ..} records enthält:
+(def btc-sma 
+  (pipe [btc [:Close]]
+    (fn [price] 
+      ; pipe mappt die Elemente des Record-Streams (hier :Close) auf die Parameter der fn. Price ist als eine ScalarSeries.
+      ; (Index=0 ist das neueste Element)
+      ; emit ist eine von pipe in den lambda-scope injizierte Funktion, die das neue Element in den Ergebnis-Stream 
+      ; pusht (und damit die Observer des Ergebnis-Streams benachrichtigt).
+      (emit {:ma (sma (get price 0))})
+    )
+  )
+)
+
+; Diese pipe hat zwei Streams als Input: den Closing-Price und den SMA. Jede pipe kann also N Input Stream haben
+(def btc-signal
+  (pipe [btc [:Close] btc-sma [:ma]]
+    (fn [price ma]
+	(emit {:signal (? (> (get price 0) (get ma 0)) :buy : sell)})
+    )
+  )
+)
+*)
+
+unit Myc.Ast.Pipes;
+
+interface
+
+uses
+    System.SysUtils,
+    System.Classes,
+    System.Generics.Collections,
+    System.SyncObjs,
+    Myc.Data.Value,
+    Myc.Data.Scalar,
+    Myc.Data.Keyword,
+    Myc.Data.Stream,
+    Myc.Core.Notifier;
+
+type
+    // Forward declaration
+    TPipeStream = class;
+
+    // mapped fields per input stream
+    TPipeConfig = TArray<TArray<TScalarRecordField>>;
+
+    TPipeSource = class(TInterfacedObject, IStreamObserver)
+    private
+        [Weak]
+        FOwner: TPipeStream;
+        FSource: IStream;
+        FTag: TSubscriptionTag;
+
+        // State for barrier synchronization
+        FLastSeenCycle: Int64;
+
+        procedure OnSignal(const Signal: TStreamSignal);
+        procedure Destroying;
+
+    public
+        constructor Create(AOwner: TPipeStream; ASource: IStream; const AInputs: TArray<TScalarRecordField>);
+        property LastSeenCycle: Int64 read FLastSeenCycle;
+    end;
+
+    // --- Runtime: The Pipe ---
+    // The executable instance of a pipe.
+    // Acts as IStream (Producer) for downstream consumers and manages internal inputs.
+    TPipeStream = class(TInterfacedObject, IStream)
+    private
+        FConfig: TPipeConfig;
+        FSources: TArray<TPipeSource>;
+        FObservers: TMycNotifyList<IStreamObserver>;
+        FSeries: IWriteableScalarRecordSeries;
+
+        // Barrier State
+        FLastFiredCycleID: Int64;
+
+        // Execution Logic
+        // The compiled lambda function representing (fn [inputs...] ...)
+        FLambda: TDataValue.TFunc;
+        FLambdaArgs: TArray<TDataValue>;
+
+        procedure CheckBarrierAndFire(CurrentCycle: Int64);
+        function GetSeries: IScalarRecordSeries;
+
+    public
+        constructor Create(const AConfig: TPipeConfig; const ASources: TArray<IStream>);
+        destructor Destroy; override;
+
+        procedure SetTransformation(const Lambda: TDataValue.TFunc);
+
+        // Injected into lambda scope as "emit"
+        procedure Emit(const Value: IScalarRecord);
+
+        // IStream Implementation
+        function Subscribe(const Observer: IStreamObserver): TSubscriptionTag;
+        procedure Unsubscribe(Tag: TSubscriptionTag);
+    end;
+
+implementation
+
+uses
+    Myc.Ast.Visitor;
+
+{ TPipeSource }
+
+constructor TPipeSource.Create(AOwner: TPipeStream; ASource: IStream; const AInputs: TArray<TScalarRecordField>);
+begin
+    inherited Create;
+
+    FOwner := AOwner;
+    FSource := ASource;
+    FLastSeenCycle := -1;
+
+    FTag := FSource.Subscribe(Self);
+end;
+
+procedure TPipeSource.Destroying;
+begin
+    FSource.Unsubscribe(FTag);
+    FOwner := nil;
+end;
+
+procedure TPipeSource.OnSignal(const Signal: TStreamSignal);
+begin
+    FLastSeenCycle := Signal.CycleID;
+
+    if Assigned(FOwner) then
+        FOwner.CheckBarrierAndFire(FLastSeenCycle);
+end;
+
+{ TPipeStream }
+
+constructor TPipeStream.Create(const AConfig: TPipeConfig; const ASources: TArray<IStream>);
+var
+    fields: TArray<TScalarRecordField>;
+begin
+    inherited Create;
+    FConfig := AConfig;
+    FLastFiredCycleID := -1;
+
+    Assert(Length(AConfig) = Length(ASources));
+
+    SetLength(FSources, Length(ASources));
+
+    // Construct record series definition from source config
+    var n := 0;
+    for var i := 0 to High(AConfig) do
+    begin
+        FSources[i] := TPipeSource.Create(Self, ASources[i], AConfig[i]);
+        FSources[i]._AddRef;
+        inc(n, Length(AConfig[i]));
+    end;
+
+    SetLength(FLambdaArgs, 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];
+
+            FLambdaArgs[n] := TDataValue.FromSeries(ASources[i].Series[j].Value);
+
+            inc(n);
+        end;
+    end;
+
+    FSeries := TScalarRecordSeries.Create(TScalarRecord.TRegistry.Intern(fields));
+end;
+
+destructor TPipeStream.Destroy;
+begin
+    for var src in FSources do
+    begin
+        src.Destroying;
+        src._Release;
+    end;
+    FObservers.Finalize;
+    inherited;
+end;
+
+procedure TPipeStream.SetTransformation(const Lambda: TDataValue.TFunc);
+begin
+    FLambda := Lambda;
+end;
+
+function TPipeStream.Subscribe(const Observer: IStreamObserver): TSubscriptionTag;
+begin
+    FObservers.Lock;
+    try
+        Result := FObservers.Advise(Observer);
+    finally
+        FObservers.Release;
+    end;
+end;
+
+procedure TPipeStream.Unsubscribe(Tag: TSubscriptionTag);
+begin
+    FObservers.Lock;
+    try
+        FObservers.Unadvise(Tag);
+    finally
+        FObservers.Release;
+    end;
+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);
+        var signal := TStreamSignal.Create(skData, FLastFiredCycleID);
+
+        FObservers.Notify(
+            function(const Obs: IStreamObserver): Boolean
+            begin
+                Obs.OnSignal(signal);
+                Result := True;
+            end
+        );
+    finally
+        FObservers.Release;
+    end;
+end;
+
+function TPipeStream.GetSeries: IScalarRecordSeries;
+begin
+    Result := FSeries;
+end;
+
+end.

Src/AST/Myc.Ast.pas

@@ -1,3 +1,6 @@
+//==================================================================================================
+//== FULL UNIT START: Myc.Ast (from Myc.Ast.pas)
+//==================================================================================================
 unit Myc.Ast;
 
 interface
@@ -327,6 +330,27 @@ type
             const AStaticType: IStaticType = nil
         ): ISeriesLengthNode; overload; static;
 
+        // --- PIPES ---
+
+        class function PipeInput(
+            const AStream: IIdentifierNode;
+            const ASelector: IKeywordNode = nil;
+            const Loc: ISourceLocation = nil
+        ): IPipeInputNode; static;
+
+        class function Pipe(
+            const AInputs: TArray<IPipeInputNode>;
+            const ATransformation: ILambdaExpressionNode;
+            const Loc: ISourceLocation = nil
+        ): IPipeNode; overload; static;
+
+        class function Pipe(
+            const Identity: IAstIdentity;
+            const AInputs: IPipeInputList;
+            const ATransformation: ILambdaExpressionNode;
+            const AStaticType: IStaticType = nil
+        ): IPipeNode; overload; static;
+
     end;
 
 implementation
@@ -909,4 +933,47 @@ begin
         );
 end;
 
+// --- PIPES ---
+
+class function TAst.PipeInput(const AStream: IIdentifierNode; const ASelector: IKeywordNode; const Loc: ISourceLocation): IPipeInputNode;
+begin
+    var id := TIdentities.Structural(Loc);
+    // Erzeugt Node aus Myc.Ast.Nodes
+    Result := TPipeInputNode.Create(AStream, ASelector, id, TTypes.Unknown);
+end;
+
+class function TAst.Pipe(
+    const AInputs: TArray<IPipeInputNode>;
+    const ATransformation: ILambdaExpressionNode;
+    const Loc: ISourceLocation
+): IPipeNode;
+begin
+    var listId := TIdentities.List('[', ']', ' ', Loc);
+    var inputList := TPipeInputList.Create(AInputs, listId);
+
+    var id := TIdentities.Structural(Loc);
+    // Erzeugt Node aus Myc.Ast.Nodes
+    Result := TPipeNode.Create(inputList, ATransformation, id, TTypes.Unknown);
+end;
+
+class function TAst.Pipe(
+    const Identity: IAstIdentity;
+    const AInputs: IPipeInputList;
+    const ATransformation: ILambdaExpressionNode;
+    const AStaticType: IStaticType
+): IPipeNode;
+begin
+    Result :=
+        TPipeNode.Create(
+            AInputs,
+            ATransformation,
+            Identity,
+            if AStaticType <> nil then AStaticType
+            else TTypes.Unknown
+        );
+end;
+
 end.
+    //==================================================================================================
+//== FULL UNIT END: Myc.Ast
+//==================================================================================================

Src/Data/Myc.Data.Scalar.pas

@@ -133,21 +133,29 @@ type
     // A field definition for a scalar record.
     TScalarRecordField = TPair<IKeyword, TScalar.TKind>;
     IScalarRecordDefinition = IKeywordMapping<TScalar.TKind>;
-    IScalarRecord = IKeywordMapping<TScalar>;
 
-    TScalarRecord = class(TInterfacedObject, IScalarRecord)
+    IScalarRecord = interface(IKeywordMapping<TScalar>)
+        {$region 'private'}
+        function GetDef: IScalarRecordDefinition;
+        {$endregion}
+        property Def: IScalarRecordDefinition read GetDef;
+    end;
+
+    TScalarRecord = class(TInterfacedObject, IKeywordMapping<TScalar>, IScalarRecord)
         type
             TRegistry = TKeywordMappingRegistry<TScalar.TKind>;
     private
-        FDef: IScalarRecordDefinition;
         FData: TArray<TScalar.TValue>;
+        FDef: IScalarRecordDefinition;
 
         function IndexOf(const Key: IKeyword): Integer;
         function GetCount: Integer;
+        function GetDef: IScalarRecordDefinition;
         function GetItems(Idx: Integer): TPair<IKeyword, TScalar>;
         function GetFields(const Key: IKeyword): TScalar;
     public
         constructor Create(const ADef: IScalarRecordDefinition; const AData: TArray<TScalar.TValue>);
+        property Def: IScalarRecordDefinition read GetDef;
     end;
 
     ISeries = interface
@@ -179,27 +187,25 @@ type
         procedure Add(const Item: TScalar.TValue; Lookback: Int64 = -1);
     end;
 
-    IScalarRecordSeries = interface
+    IScalarRecordSeries = interface(IKeywordMapping<ISeries>)
         {$region 'private'}
-        function GetCount: Int64;
         function GetDef: IScalarRecordDefinition;
-        function GetFields(const Key: IKeyword): ISeries;
-        function GetTotalCount: Int64;
         {$endregion}
-        property Count: Int64 read GetCount;
         property Def: IScalarRecordDefinition read GetDef;
-        property Fields[const Key: IKeyword]: ISeries read GetFields; default;
-        property TotalCount: Int64 read GetTotalCount;
     end;
 
     IWriteableScalarRecordSeries = interface(IScalarRecordSeries)
         {$region 'private'}
+        function GetRecordCount: Int64;
+        function GetTotalCount: Int64;
         {$endregion}
-        procedure Add(const Item: IScalarRecord; Lookback: Int64 = -1);
+        procedure Add(const Item: IKeywordMapping<TScalar>; Lookback: Int64 = -1);
+        property RecordCount: Int64 read GetRecordCount;
+        property TotalCount: Int64 read GetTotalCount;
     end;
 
     // A series of scalar records, optimized for memory and access speed.
-    TScalarRecordSeries = class(TInterfacedObject, IWriteableScalarRecordSeries)
+    TScalarRecordSeries = class(TInterfacedObject, IWriteableScalarRecordSeries, IKeywordMapping<ISeries>)
         type
             TMemberSeries = class(TGenericContainedObject<TScalarRecordSeries>, ISeries)
             private
@@ -218,19 +224,24 @@ type
         FArray: TChunkArray<TScalar.TValue>;
         FFields: TArray<TMemberSeries>;
         FTotalCount: Int64;
-        function GetCount: Int64; inline;
+
+        function GetItems(Idx: Integer): TPair<IKeyword, ISeries>;
+        function GetCount: Integer;
+        function IndexOf(const Key: IKeyword): Integer;
+
+        function GetRecordCount: Int64; inline;
         function GetFields(const Key: IKeyword): ISeries;
         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: IScalarRecord; Lookback: Int64 = -1);
+        procedure Add(const Item: IKeywordMapping<TScalar>; Lookback: Int64 = -1);
 
-        property Count: Int64 read GetCount;
-        property Fields[const Key: IKeyword]: ISeries read GetFields; default;
+        property RecordCount: Int64 read GetRecordCount;
         property Def: IScalarRecordDefinition read GetDef;
         property TotalCount: Int64 read GetTotalCount;
     end;
@@ -803,7 +814,7 @@ begin
     inherited;
 end;
 
-procedure TScalarRecordSeries.Add(const Item: IScalarRecord; Lookback: Int64 = -1);
+procedure TScalarRecordSeries.Add(const Item: IKeywordMapping<TScalar>; Lookback: Int64 = -1);
 begin
     var lb := FDef.Count * Integer(Lookback);
     for var i := 0 to FDef.Count - 1 do
@@ -820,7 +831,7 @@ begin
     Result := FFields[elem];
 end;
 
-function TScalarRecordSeries.GetCount: Int64;
+function TScalarRecordSeries.GetRecordCount: Int64;
 begin
     Result := FArray.Count div FDef.Count;
 end;
@@ -836,11 +847,32 @@ 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;
 end;
 
+function TScalarRecordSeries.GetCount: Integer;
+begin
+    Result := Length(FFields);
+end;
+
+function TScalarRecordSeries.GetItems(Idx: Integer): TPair<IKeyword, ISeries>;
+begin
+    Result.Key := FDef.Items[Idx].Key;
+    Result.Value := FFields[Idx];
+end;
+
+function TScalarRecordSeries.IndexOf(const Key: IKeyword): Integer;
+begin
+    Result := FDef.IndexOf(Key);
+end;
+
 { TScalar.TKindHelper }
 
 function TScalar.TKindHelper.ToString: string;
@@ -907,7 +939,7 @@ end;
 
 function TScalarRecordSeries.TMemberSeries.GetCount: Int64;
 begin
-    Result := RecordSeries.Count;
+    Result := RecordSeries.RecordCount;
 end;
 
 function TScalarRecordSeries.TMemberSeries.GetItems(Idx: Integer): TScalar;
@@ -998,6 +1030,11 @@ begin
     Result := FDef.Count;
 end;
 
+function TScalarRecord.GetDef: IScalarRecordDefinition;
+begin
+    Result := FDef;
+end;
+
 function TScalarRecord.GetFields(const Key: IKeyword): TScalar;
 var
     idx: Integer;

Src/Data/Myc.Data.Stream.pas

@@ -0,0 +1,268 @@
+unit Myc.Data.Stream;
+
+interface
+
+uses
+    System.SysUtils,
+    System.Classes,
+    System.Generics.Collections,
+    System.SyncObjs,
+    Myc.Data.Scalar,
+    Myc.Data.Value,
+    Myc.Core.Notifier;
+
+type
+    TSignalKind = (skData, skHeartbeat);
+
+    TStreamSignal = record
+    private
+        FCycleID: Int64;
+        FKind: TSignalKind;
+    public
+        constructor Create(AKind: TSignalKind; ACycleID: Int64);
+        function ToString: string;
+        property CycleID: Int64 read FCycleID;
+        property Kind: TSignalKind read FKind;
+    end;
+
+    IStreamObserver = interface
+        procedure OnSignal(const Signal: TStreamSignal);
+    end;
+
+    TSubscriptionTag = Pointer;
+
+    IStream = interface
+        {$region 'private'}
+        function GetSeries: IScalarRecordSeries;
+        {$endregion}
+
+        function Subscribe(const Observer: IStreamObserver): TSubscriptionTag;
+        procedure Unsubscribe(Tag: TSubscriptionTag);
+
+        // Returns the Series of the records this stream produces. Streams are scalar record series. Always!
+        property Series: IScalarRecordSeries read GetSeries;
+    end;
+
+    IInputChannel = interface
+        {$region 'private'}
+        function GetStream: IStream;
+        {$endregion}
+
+        procedure Push(const Value: IScalarRecord);
+        property Stream: IStream read GetStream;
+    end;
+
+    IGraphExecutor = interface
+        {$region 'private'}
+        function GetCycleID: Int64;
+        {$endregion}
+
+        // Now requires a Definition when creating a channel
+        function GetInputChannel(const Name: string; const Def: IScalarRecordDefinition): IInputChannel;
+        procedure Step;
+        property CycleID: Int64 read GetCycleID;
+    end;
+
+    TGraphExecutor = class(TInterfacedObject, IGraphExecutor)
+    private
+        type
+            TChannel = class(TInterfacedObject, IInputChannel, IStream)
+            private
+                type
+                    TStreamNotifier = TMycNotifyList<IStreamObserver>;
+            private
+                FName: string;
+                FSeries: IWriteableScalarRecordSeries;
+                FQueue: TQueue<IScalarRecord>;
+                FQueueLock: TSpinLock;
+                FObservers: TStreamNotifier;
+
+                procedure Push(const Value: IScalarRecord);
+                function GetStream: IStream;
+                function Subscribe(const Observer: IStreamObserver): TSubscriptionTag;
+                procedure Unsubscribe(Tag: TSubscriptionTag);
+                procedure Emit(ACycle: Int64);
+
+                function GetSeries: IScalarRecordSeries;
+
+            public
+                constructor Create(const AName: string; const ASeries: IWriteableScalarRecordSeries);
+                destructor Destroy; override;
+            end;
+
+    private
+        FCycleID: Int64;
+        FChannels: TDictionary<string, IInputChannel>;
+        FLock: TSpinLock;
+        function GetInputChannel(const Name: string; const Def: IScalarRecordDefinition): IInputChannel;
+        function GetCycleID: Int64;
+    public
+        constructor Create;
+        destructor Destroy; override;
+        procedure Step;
+    end;
+
+implementation
+
+constructor TStreamSignal.Create(AKind: TSignalKind; ACycleID: Int64);
+begin
+    FKind := AKind;
+    FCycleID := ACycleID;
+end;
+
+function TStreamSignal.ToString: string;
+begin
+    if Kind = skData then
+        Result := Format('Signal(Data, #%d)', [CycleID])
+    else
+        Result := Format('Signal(Heartbeat, #%d)', [CycleID]);
+end;
+
+{ TGraphExecutor.TChannel }
+
+constructor TGraphExecutor.TChannel.Create(const AName: string; const ASeries: IWriteableScalarRecordSeries);
+begin
+    inherited Create;
+    FName := AName;
+    FSeries := ASeries;
+    FQueue := TQueue<IScalarRecord>.Create;
+end;
+
+destructor TGraphExecutor.TChannel.Destroy;
+begin
+    FObservers.Finalize;
+    FQueue.Free;
+    inherited;
+end;
+
+procedure TGraphExecutor.TChannel.Push(const Value: IScalarRecord);
+begin
+    Assert(Value.Def = FSeries.Def);
+
+    FQueueLock.Enter;
+    try
+        FQueue.Enqueue(Value);
+    finally
+        FQueueLock.Exit;
+    end;
+end;
+
+function TGraphExecutor.TChannel.GetStream: IStream;
+begin
+    Result := Self;
+end;
+
+function TGraphExecutor.TChannel.Subscribe(const Observer: IStreamObserver): TSubscriptionTag;
+begin
+    FObservers.Lock;
+    try
+        Result := FObservers.Advise(Observer);
+    finally
+        FObservers.Release;
+    end;
+end;
+
+procedure TGraphExecutor.TChannel.Unsubscribe(Tag: TSubscriptionTag);
+begin
+    FObservers.Lock;
+    try
+        FObservers.Unadvise(Tag);
+    finally
+        FObservers.Release;
+    end;
+end;
+
+procedure TGraphExecutor.TChannel.Emit(ACycle: Int64);
+var
+    signal: TStreamSignal;
+begin
+    FQueueLock.Enter;
+    try
+        FSeries.Add(FQueue.Dequeue);
+        signal :=
+            TStreamSignal.Create(
+                if FQueue.Count > 0 then skData
+                else skHeartbeat,
+                ACycle
+            )
+    finally
+        FQueueLock.Exit;
+    end;
+
+    FObservers.Lock;
+    try
+        FObservers.Notify(
+            function(const Obs: IStreamObserver): Boolean
+            begin
+                Obs.OnSignal(signal);
+                Result := True;
+            end
+        );
+    finally
+        FObservers.Release;
+    end;
+end;
+
+function TGraphExecutor.TChannel.GetSeries: IScalarRecordSeries;
+begin
+    Result := FSeries;
+end;
+
+{ TGraphExecutor }
+
+constructor TGraphExecutor.Create;
+begin
+    inherited Create;
+    FChannels := TDictionary<string, IInputChannel>.Create;
+    FCycleID := 0;
+end;
+
+destructor TGraphExecutor.Destroy;
+begin
+    FChannels.Free;
+    inherited;
+end;
+
+function TGraphExecutor.GetInputChannel(const Name: string; const Def: IScalarRecordDefinition): IInputChannel;
+var
+    impl: TChannel;
+begin
+    FLock.Enter;
+    try
+        if not FChannels.TryGetValue(Name, Result) then
+        begin
+            impl := TChannel.Create(Name, TScalarRecordSeries.Create(Def));
+            Result := impl;
+            FChannels.Add(Name, Result);
+        end;
+    finally
+        FLock.Exit;
+    end;
+end;
+
+function TGraphExecutor.GetCycleID: Int64;
+begin
+    Result := FCycleID;
+end;
+
+procedure TGraphExecutor.Step;
+var
+    channelsArr: TArray<IInputChannel>;
+    channel: IInputChannel;
+begin
+    FLock.Enter;
+    try
+        Inc(FCycleID);
+        channelsArr := FChannels.Values.ToArray;
+    finally
+        FLock.Exit;
+    end;
+
+    for channel in channelsArr do
+        (channel as TChannel).Emit(FCycleID);
+end;
+
+initialization
+    TMycNotifyList<IStreamObserver>.ReverseOnNotify := False;
+
+end.

Src/Data/Myc.Data.Value.pas

@@ -397,7 +397,7 @@ begin
                     first := False;
                 end;
                 sb.Append('}');
-                Result := Format('<record_series%s[%d]>', [sb.ToString, series.Count]);
+                Result := Format('<record_series%s[%d]>', [sb.ToString, series.RecordCount]);
             finally
                 sb.Free;
             end;