Memory hole in Data-Join fixed

AuraTrader/MainForm.pas

@@ -315,15 +315,6 @@ begin
 
     // next stage
 
-    var ATREndPoint := ATR.CreateEndpoint(5);
-    var ATRSeries: TSeries<Double>;
-
-    var HullEndPoint := Hull.CreateEndpoint(5);
-    var HullSeries: TSeries<Double>;
-
-    var SmaEndPoint := Sma.CreateEndpoint(5);
-    var SmaSeries: TSeries<Double>;
-
     var curr: TSignal;
     curr.SL := Double.NaN;
     curr.Entry := Double.NaN;
@@ -333,7 +324,7 @@ begin
     var conv := TConverter.Join<Double>([Ohlc.Field<Double>('Low'), Ohlc.Field<Double>('High'), Closes, ATR, Hull, Sma]);
 
     var Signal :=
-        TConverter<TArray<Double>, TSignal>.CreateGeneric(
+        TConverter<TArray<Double>, TSignal>.CreateConverter(
             function(const Values: TArray<Double>): TSignal
             begin
                 var low := Values[0];
@@ -473,7 +464,7 @@ begin
     panel.AddDoubleSeries(Signal.Producer.Field<Double>('Entry'), TAlphaColors.Green, 1);
     panel.AddDoubleSeries(Signal.Producer.Field<Double>('SL'), TAlphaColors.Red, 2);
 
-    var mean := TConverter<TArray<Double>, Double>.CreateGeneric(TIndicators.CreateMean());
+    var mean := TConverter<TArray<Double>, Double>.CreateConverter(TIndicators.CreateMean());
 
     TConverter.Join<Double>([Hull, Sma]).Chain<Double>(mean);
 
@@ -494,14 +485,14 @@ begin
 
     var indi := EMAFActory.CreateIndicator(Params);
 
-    var EMAConv := TConverter<TDataRecord, TDataRecord>.CreateGeneric(indi);
+    var EMAConv := TConverter<TDataRecord, TDataRecord>.CreateConverter(indi);
 
     var equityEMA :=
         equity
             .Producer
-            .Chain<TDataRecord>(TConverter.FieldToRecord<Double>(EMAFactory.Input, 'Price'))
+            .Chain<TDataRecord>(EMAFactory.Input.FieldAsRecord<Double>('Price'))
             .Chain<TDataRecord>(EMAConv)
-            .Chain<Double>(TConverter.FieldOfRecord<Double>(EMAFactory.Output, 'MA'));
+            .Chain<Double>(EMAFactory.Output.FieldOfRecord<Double>('MA'));
 
     //////////////
 

AuraTrader/StrategyTest.pas

@@ -50,7 +50,7 @@ begin
 
     var signalGenerator :=
         conv.Chain<TSignalEvent>(
-            TConverter<TArray<Double>, TSignalEvent>.CreateGeneric(
+            TConverter<TArray<Double>, TSignalEvent>.CreateConverter(
                 function(const Values: TArray<Double>): TSignalEvent
                 begin
                     Result.Low := Values[0];

Src/Myc.Fmx.Chart.Series.pas

@@ -18,8 +18,8 @@ type
     { TChartCustomLayer }
     TChartCustomLayer<T> = class abstract(TMycChart.TDataLayer)
     private
-        FProducer: TProducer<T>;
+        FProducer: TProducer<T>.TSubscription;
-        FReceiver: TLazy<TSeries<T>>;
+        FSeries: TLazy<TSeries<T>>;
         FData: TSeries<T>;
     protected
         function GetCount: Int64; override;
@@ -34,6 +34,7 @@ type
         ); override; abstract;
     public
         constructor Create(AParent: TMycChart.TPanel; const AProducer: TProducer<T>);
+        destructor Destroy; override;
         property Data: TSeries<T> read FData;
     end;
 
@@ -83,7 +84,7 @@ type
     { TChartXAxisLayer }
     TChartXAxisLayer<T> = class(TMycChart.TXAxisLayer)
     private
-        FProducer: TProducer<T>;
+        FProducer: TProducer<T>.TSubscription;
         FReceiver: TLazy<TSeries<T>>;
         FData: TSeries<T>;
     protected
@@ -93,6 +94,7 @@ type
         function GetTotalCount: Int64; override;
     public
         constructor Create(AOwner: TMycChart; const AProducer: TProducer<T>);
+        destructor Destroy; override;
         property Data: TSeries<T> read FData;
     end;
 
@@ -270,9 +272,13 @@ end;
 constructor TChartCustomLayer<T>.Create(AParent: TMycChart.TPanel; const AProducer: TProducer<T>);
 begin
     inherited Create(AParent);
-    FProducer := AProducer;
+    FProducer := AProducer.CreateEndpoint(Owner.Lookback.Value, FSeries);
+end;
 
-    FReceiver := FProducer.CreateEndpoint(Owner.Lookback.Value);
+destructor TChartCustomLayer<T>.Destroy;
+begin
+    FProducer.Unlink;
+    inherited;
 end;
 
 function TChartCustomLayer<T>.GetCount: Int64;
@@ -287,7 +293,7 @@ end;
 
 function TChartCustomLayer<T>.Update: Boolean;
 begin
-    Result := FReceiver.Update(FData);
+    Result := FSeries.Update(FData);
 end;
 
 { TChartXAxisLayer }
@@ -295,8 +301,13 @@ end;
 constructor TChartXAxisLayer<T>.Create(AOwner: TMycChart; const AProducer: TProducer<T>);
 begin
     inherited Create(AOwner);
-    FProducer := AProducer;
+    FProducer := AProducer.CreateEndpoint(Owner.Lookback.Value, FReceiver);
-    FReceiver := FProducer.CreateEndpoint(Owner.Lookback.Value);
+end;
+
+destructor TChartXAxisLayer<T>.Destroy;
+begin
+    FProducer.Unlink;
+    inherited;
 end;
 
 function TChartXAxisLayer<T>.GetCaption(Idx: Int64): String;

Src/Myc.Trade.DataPoint.Impl.pas

@@ -31,6 +31,16 @@ type
         class property Null: IConsumer<T> read FNull;
     end;
 
+    // A consumer implementation that is owned by a controller.
+    TMycGenericConsumer<T> = class(TMycConsumer<T>)
+    private
+        FProc: TConstFunc<T, TState>;
+    protected
+        function Consume(const Value: T): TState; override; final;
+    public
+        constructor Create(const Controller: IInterface; const AProc: TConstFunc<T, TState>);
+    end;
+
     TMycProducer<T> = class(TInterfacedObject, IProducer<T>)
     strict private
         type
@@ -79,18 +89,7 @@ type
             FNull: IConverter<S, T>;
         class constructor CreateClass;
     private
-        type
+        FConsumer: TMycGenericConsumer<S>;
-            TConsumer = class(TMycConsumer<S>)
-            private
-                FOwner: TMycConverter<S, T>;
-            protected
-                function Consume(const Value: S): TState; override; final;
-            public
-                constructor Create(AOwner: TMycConverter<S, T>);
-            end;
-
-    private
-        FConsumer: TConsumer;
         function GetConsumer: IConsumer<S>;
     protected
         // To be implemented by descendants to perform the actual conversion.
@@ -162,16 +161,6 @@ type
         constructor Create(const AFieldName: String);
     end;
 
-    // A consumer implementation that is owned by a controller.
-    TMycGenericConsumer<T> = class(TMycConsumer<T>)
-    private
-        FProc: TConstFunc<T, TState>;
-    protected
-        function Consume(const Value: T): TState; override;
-    public
-        constructor Create(const Controller: IInterface; const AProc: TConstFunc<T, TState>);
-    end;
-
     // Endpoint that collects data into a series.
     TMycDataEndpoint<T> = class(TInterfacedObject, TLazy<TSeries<T>>.ILazy)
     private
@@ -183,9 +172,7 @@ type
             end;
     private
         FConsumer: TMycGenericConsumer<T>;
-        FTag: TTag;
+        FLookback: Integer;
-        FProducer: IProducer<T>;
-        FLookback: Int64;
         FChanged: TFlag;
         FLock: TLightweightMREW;
         FFirst: PItem;
@@ -193,9 +180,11 @@ type
         function GetChanged: TState;
         function Consume(const Value: T): TState;
     public
-        constructor Create(const AProducer: TProducer<T>; ALookback: Int64);
+        constructor Create(ALookback: Integer);
         destructor Destroy; override;
         function Update(var Value: TSeries<T>): Boolean;
+
+        class function CreateDataEndpoint(Lookback: Integer; out Series: TLazy<TSeries<T>>): IConsumer<T>;
     end;
 
     TTickAggregation = class(TMycConverter<TDataPoint<Double>, TDataPoint<TOhlcItem>>)
@@ -239,9 +228,7 @@ type
     TMycDataJoin<T> = class(TInterfacedObject, IProducer<TArray<T>>)
     private
         FConsumers: array of record
-            Producer: IProducer<T>;
             Consumer: TMycGenericConsumer<T>;
-            Tag: TTag;
             Queue: TQueue<T>;
         end;
 
@@ -249,11 +236,13 @@ type
         FLock: TSpinLock;
         FCount: Integer;
         function Consume(Idx: Integer; const Value: T): TState;
+        function GetConsumers(Idx: Integer): IConsumer<T>;
         function Link(const Consumer: IConsumer<TArray<T>>): TTag;
         procedure Unlink(Tag: TTag);
     public
-        constructor Create(const AProducers: TArray<TProducer<T>>);
+        constructor Create(ACount: Integer);
         destructor Destroy; override;
+        property Consumers[Idx: Integer]: IConsumer<T> read GetConsumers;
     end;
 
     TMycComposedConverter<S, T> = class(TInterfacedObject, IConverter<S, T>)
@@ -363,7 +352,7 @@ end;
 constructor TMycConverter<S, T>.Create;
 begin
     inherited Create;
-    FConsumer := TConsumer.Create(Self);
+    FConsumer := TMycGenericConsumer<S>.Create(Self, Consume);
 end;
 
 { TMycConverter<S, T> }
@@ -486,19 +475,15 @@ end;
 
 { TMycDataEndpoint<T> }
 
-constructor TMycDataEndpoint<T>.Create(const AProducer: TProducer<T>; ALookback: Int64);
+constructor TMycDataEndpoint<T>.Create(ALookback: Integer);
 begin
     inherited Create;
-    FProducer := AProducer;
     FLookback := ALookback;
-
     FConsumer := TMycGenericConsumer<T>.Create(Self, Consume);
-    FTag := FProducer.Link(FConsumer);
 end;
 
 destructor TMycDataEndpoint<T>.Destroy;
 begin
-    FProducer.Unlink(FTag);
     FConsumer.Free;
     inherited;
 end;
@@ -524,6 +509,15 @@ begin
     end;
 end;
 
+class function TMycDataEndpoint<T>.CreateDataEndpoint(Lookback: Integer; out Series: TLazy<TSeries<T>>): IConsumer<T>;
+var
+    endPoint: TMycDataEndpoint<T>;
+begin
+    endPoint := TMycDataEndpoint<T>.Create(Lookback);
+    Result := endPoint.FConsumer;
+    Series := endPoint;
+end;
+
 function TMycDataEndpoint<T>.Update(var Value: TSeries<T>): Boolean;
 begin
     FLock.BeginWrite;
@@ -801,11 +795,11 @@ end;
 
 { TMycDataJoin<T> }
 
-constructor TMycDataJoin<T>.Create(const AProducers: TArray<TProducer<T>>);
+constructor TMycDataJoin<T>.Create(ACount: Integer);
 begin
     inherited Create;
 
-    SetLength(FConsumers, Length(AProducers));
+    SetLength(FConsumers, ACount);
 
     FLock := TSpinLock.Create(false);
     FCount := Length(FConsumers);
@@ -822,9 +816,7 @@ begin
         with FConsumers[i] do
         begin
             Queue := TQueue<T>.Create;
-            Producer := AProducers[i];
             Consumer := TMycGenericConsumer<T>.Create(Self, cFunc(i));
-            Tag := Producer.Link(Consumer);
         end;
 end;
 
@@ -833,7 +825,6 @@ begin
     for var i := High(FConsumers) downto 0 do
         with FConsumers[i] do
         begin
-            Producer.Unlink(FConsumers[i].Tag);
             Consumer.Free;
             Queue.Free;
         end;
@@ -884,6 +875,11 @@ begin
         FContainedProvider.Broadcast(Arr);
 end;
 
+function TMycDataJoin<T>.GetConsumers(Idx: Integer): IConsumer<T>;
+begin
+    Result := FConsumers[Idx].Consumer;
+end;
+
 { TMycGenericAggregator<S, T> }
 
 constructor TMycGenericAggregator<S, T>.Create(const AFunc: TConstFunc<S, TConverter<S, T>.TBroadcastProc, TState>);
@@ -1005,15 +1001,4 @@ begin
 
 end;
 
-constructor TMycConverter<S, T>.TConsumer.Create(AOwner: TMycConverter<S, T>);
-begin
-    inherited Create(AOwner);
-    FOwner := AOwner;
-end;
-
-function TMycConverter<S, T>.TConsumer.Consume(const Value: S): TState;
-begin
-    Result := FOwner.Consume(Value);
-end;
-
 end.

Src/Myc.Trade.DataPoint.pas

@@ -17,6 +17,7 @@ type
         function Consume(const Value: T): TState;
     end;
 
+    // A producer generates data and distributes it to linked consumers.
     IProducer<T> = interface
         function Link(const Consumer: IConsumer<T>): TTag;
         procedure Unlink(Tag: TTag);
@@ -30,17 +31,18 @@ type
         property Consumer: IConsumer<S> read GetConsumer;
     end;
 
-    // Interface helper for IProducer providing the null object pattern.
+    // Interface helper for IProducer providing the null object pattern nad subscriptions for linked consumers.
     TProducer<T> = record
     public
         type
             TSubscription = record
             private
-                FProducer: IProducer<T>;
+                FOwner: IProducer<T>;
                 FTag: TTag;
             public
                 procedure Unlink;
-                property Producer: IProducer<T> read FProducer;
+                class operator Implicit(A: TSubscription): TProducer<T>; overload;
+                property Owner: IProducer<T> read FOwner;
             end;
     private
         FProducer: IProducer<T>;
@@ -54,17 +56,14 @@ type
         class operator Implicit(const A: IProducer<T>): TProducer<T>; overload;
         class operator Implicit(const A: TProducer<T>): IProducer<T>; overload;
 
-        function Link(const Consumer: IConsumer<T>): TSubscription; inline;
+        function CreateLink(const Consumer: IConsumer<T>): TSubscription; inline;
+        function CreateEndpoint(Lookback: Int64; out Series: TLazy<TSeries<T>>): TSubscription;
+        function CreateSequence(Count: Integer; out Seq: TArray<IProducer<T>>): TSubscription; overload; experimental;
 
+        // Chain consumers
         function Chain(const Next: IConsumer<T>): IConsumer<T>; overload; inline;
-
         function Chain<R>(const Next: IConverter<T, R>): TProducer<R>; overload; inline;
         function Chain<R>(const Func: TConstFunc<T, R>): TProducer<R>; overload; inline;
-
-        function CreateEndpoint(Lookback: Int64): TLazy<TSeries<T>>;
-
-        function CreateSequence(Count: Integer): TArray<IProducer<T>>; overload; experimental;
-
         // Extracts the field of a record by it's name (using RTTI).
         function Field<R>(const FieldName: String): TProducer<R>; inline;
 
@@ -93,7 +92,8 @@ type
         class operator Implicit(const A: TConverter<S, T>): IConverter<S, T>; overload;
 
         class function Construct(const Consumer: IConsumer<S>; const Producer: TProducer<T>): TConverter<S, T>; static;
-        class function CreateGeneric(const Func: TConstFunc<S, T>): TConverter<S, T>; static;
+
+        class function CreateConverter(const Func: TConstFunc<S, T>): TConverter<S, T>; static;
         class function CreateAggregation(const Func: TConstFunc<S, TBroadcastProc, TState>): TConverter<S, T>; static;
 
         property Consumer: IConsumer<S> read GetConsumer;
@@ -102,20 +102,21 @@ type
         class property Null: IConverter<S, T> read GetNull;
     end;
 
+    TDataRecordLayoutHelper = record helper for TDataRecord.TLayout
+        function FieldAsRecord<T>(const Name: String): TConverter<T, TDataRecord>;
+        function FieldOfRecord<T>(const Name: String): TConverter<TDataRecord, T>;
+    end;
+
     // Factory for creating specific converter instances.
     TConverter = record
         class function CreateCounter<T>: TConverter<T, Int64>; static;
         class function CreateTicker<T>: TConverter<TArray<T>, T>; static;
-        class function CreateRecordField<S, T>(const FieldName: String): TConverter<S, T>; static;
         class function CreateIdentity<T>: TConverter<T, T>; static;
 
         class function CreateTickAggregation(Timeframe: TTimeframe): TConverter<TDataPoint<Double>, TDataPoint<TOhlcItem>>; static;
         class function CreateOhlcAggregation(Timeframe: TTimeframe): TConverter<TDataPoint<TOhlcItem>, TDataPoint<TOhlcItem>>; static;
 
-        class function CreateDataPointConverter<S, T>(const Func: TConstFunc<S, T>): TConverter<TDataPoint<S>, TDataPoint<T>>; static;
+        class function CreateEndpoint<T>(Lookback: Int64; out Series: TLazy<TSeries<T>>): IConsumer<T>; static;
-
-        class function FieldToRecord<T>(const Layout: TDataRecord.TLayout; const Name: String): TConverter<T, TDataRecord>; static;
-        class function FieldOfRecord<T>(const Layout: TDataRecord.TLayout; const Name: String): TConverter<TDataRecord, T>; static;
 
         class function Join<T>(const Producers: TArray<TProducer<T>>): TProducer<TArray<T>>; static;
 
@@ -139,6 +140,13 @@ type
         ): TConverter<TArray<TDataRecord>, TDataRecord>; static;
     end;
 
+    TDataRecordHelper = record helper for TArray<TProducer<TDataRecord>>
+        function JoinRecords(
+            const TargetLayout: TDataRecord.TLayout;
+            const Mapping: TArray<TConverter.TRecordMapping>
+        ): TProducer<TDataRecord>; experimental;
+    end;
+
 implementation
 
 uses
@@ -150,11 +158,16 @@ procedure TProducer<T>.TSubscription.Unlink;
 begin
     if FTag <> nil then
     begin
-        FProducer.Unlink(FTag);
+        FOwner.Unlink(FTag);
         FTag := nil;
     end;
 end;
 
+class operator TProducer<T>.TSubscription.Implicit(A: TSubscription): TProducer<T>;
+begin
+    Result := A.FOwner;
+end;
+
 constructor TProducer<T>.Create(const AProducer: IProducer<T>);
 begin
     FProducer := AProducer;
@@ -170,7 +183,6 @@ end;
 
 function TProducer<T>.Chain<R>(const Next: IConverter<T, R>): TProducer<R>;
 begin
-    // REFACTOR: Link to the consumer part, return the converter as the new producer
     FProducer.Link(Next.Consumer);
     Result := Next;
 end;
@@ -180,14 +192,14 @@ begin
     Result := Chain<R>(TMycGenericConverter<T, R>.Create(Func));
 end;
 
-function TProducer<T>.CreateEndpoint(Lookback: Int64): TLazy<TSeries<T>>;
+function TProducer<T>.CreateEndpoint(Lookback: Int64; out Series: TLazy<TSeries<T>>): TSubscription;
 begin
-    Result := TMycDataEndpoint<T>.Create(FProducer, Lookback);
+    Result := CreateLink(TMycDataEndpoint<T>.CreateDataEndpoint(Lookback, Series));
 end;
 
-function TProducer<T>.CreateSequence(Count: Integer): TArray<IProducer<T>>;
+function TProducer<T>.CreateSequence(Count: Integer; out Seq: TArray<IProducer<T>>): TSubscription;
 begin
-    FProducer.Link(TMycSequence<T>.CreateSequence(Count, Result));
+    Result := CreateLink(TMycSequence<T>.CreateSequence(Count, Seq));
 end;
 
 function TProducer<T>.Field<R>(const FieldName: String): TProducer<R>;
@@ -215,9 +227,9 @@ begin
     Result := A.FProducer;
 end;
 
-function TProducer<T>.Link(const Consumer: IConsumer<T>): TSubscription;
+function TProducer<T>.CreateLink(const Consumer: IConsumer<T>): TSubscription;
 begin
-    Result.FProducer := FProducer;
+    Result.FOwner := FProducer;
     Result.FTag := FProducer.Link(Consumer);
 end;
 
@@ -243,7 +255,7 @@ begin
     Result := TMycGenericAggregator<S, T>.Create(Func);
 end;
 
-class function TConverter<S, T>.CreateGeneric(const Func: TConstFunc<S, T>): TConverter<S, T>;
+class function TConverter<S, T>.CreateConverter(const Func: TConstFunc<S, T>): TConverter<S, T>;
 begin
     Result := TMycGenericConverter<S, T>.Create(Func);
 end;
@@ -290,17 +302,9 @@ begin
     Result := TMycDataCounter<T>.Create;
 end;
 
-class function TConverter.CreateDataPointConverter<S, T>(const Func: TConstFunc<S, T>): TConverter<TDataPoint<S>, TDataPoint<T>>;
+class function TConverter.CreateEndpoint<T>(Lookback: Int64; out Series: TLazy<TSeries<T>>): IConsumer<T>;
 begin
-    var cFunc: TConstFunc<S, T> := Func;
+    Result := TMycDataEndpoint<T>.CreateDataEndpoint(Lookback, Series);
-    Result :=
-        TMycGenericConverter<TDataPoint<S>, TDataPoint<T>>.Create(
-            function(const Value: TDataPoint<S>): TDataPoint<T>
-            begin
-                Result.Time := Value.Time;
-                Result.Data := cFunc(Value.Data);
-            end
-        );
 end;
 
 class function TConverter.CreateIdentity<T>: TConverter<T, T>;
@@ -308,11 +312,6 @@ begin
     Result := TMycIdentityConverter<T>.Create;
 end;
 
-class function TConverter.CreateRecordField<S, T>(const FieldName: String): TConverter<S, T>;
-begin
-    Result := TMycRecordFieldReader<S, T>.Create(FieldName);
-end;
-
 class function TConverter.CreateTicker<T>: TConverter<TArray<T>, T>;
 begin
     Result := TMycTicker<T>.Create;
@@ -344,7 +343,7 @@ begin
     end;
 
     Result :=
-        TConverter<TArray<TDataRecord>, TDataRecord>.CreateGeneric(
+        TConverter<TArray<TDataRecord>, TDataRecord>.CreateConverter(
             function(const Inputs: TArray<TDataRecord>): TDataRecord
             begin
                 Result := TDataRecord.Create(Output);
@@ -355,28 +354,15 @@ begin
         );
 end;
 
-class function TConverter.FieldOfRecord<T>(const Layout: TDataRecord.TLayout; const Name: String): TConverter<TDataRecord, T>;
-begin
-    var idx := Layout.IndexOf(Name);
-    Result := TConverter<TDataRecord, T>.CreateGeneric(function(const Value: TDataRecord): T begin Value.GetValue(idx, Result); end);
-end;
-
-class function TConverter.FieldToRecord<T>(const Layout: TDataRecord.TLayout; const Name: String): TConverter<T, TDataRecord>;
-begin
-    var idx := Layout.IndexOf(Name);
-    Result :=
-        TConverter<T, TDataRecord>.CreateGeneric(
-            function(const Value: T): TDataRecord
-            begin
-                Result := TDataRecord.Create(Layout);
-                Result.SetValue(idx, Value);
-            end
-        );
-end;
-
 class function TConverter.Join<T>(const Producers: TArray<TProducer<T>>): TProducer<TArray<T>>;
+var
+    joiner: TMycDataJoin<T>;
 begin
-    Result := TMycDataJoin<T>.Create(Producers);
+    joiner := TMycDataJoin<T>.Create(Length(Producers));
+    Result := joiner;
+
+    for var i := 0 to High(Producers) do
+        Producers[i].Chain(joiner.Consumers[i]);
 end;
 
 class function TConverter.JoinRecords(
@@ -389,8 +375,39 @@ begin
 
     Result := DataMapping(Mapping, TargetLayout);
 
-    // REFACTOR: Link to the consumer part of the mapping converter.
+    RecProvider.Chain(Result.Consumer);
-    RecProvider.Link(Result.Consumer);
+end;
+
+function TDataRecordLayoutHelper.FieldAsRecord<T>(const Name: String): TConverter<T, TDataRecord>;
+begin
+    var layout := Self;
+    var idx := IndexOf(Name);
+    Result :=
+        TConverter<T, TDataRecord>.CreateConverter(
+            function(const Value: T): TDataRecord
+            begin
+                Result := TDataRecord.Create(layout);
+                Result.SetValue(idx, Value);
+            end
+        );
+end;
+
+function TDataRecordLayoutHelper.FieldOfRecord<T>(const Name: String): TConverter<TDataRecord, T>;
+begin
+    var idx := IndexOf(Name);
+    Result := TConverter<TDataRecord, T>.CreateConverter(function(const Value: TDataRecord): T begin Value.GetValue(idx, Result); end);
+end;
+
+function TDataRecordHelper.JoinRecords(
+    const TargetLayout: TDataRecord.TLayout;
+    const Mapping: TArray<TConverter.TRecordMapping>
+): TProducer<TDataRecord>;
+begin
+    var map := TConverter.DataMapping(Mapping, TargetLayout);
+
+    TConverter.Join<TDataRecord>(Self).Chain(map.Consumer);
+
+    Result := map.Producer;
 end;
 
 end.