Refactoring DataFlow, 1st Generic Data records working

Src/Myc.Trade.DataPoint.Impl.pas

@@ -5,6 +5,7 @@ interface
 uses
     System.SysUtils,
     System.SyncObjs,
+    System.Generics.Collections,
     Myc.Signals,
     Myc.Mutable,
     Myc.Core.Notifier,
@@ -147,13 +148,11 @@ type
 
     // A processor implementation that is owned by a controller.
     TMycContainedProcessor<T> = class(TContainedObject, IMycProcessor<T>)
-        type
-            TProc = function(const Value: T): TState of object;
     private
-        FProc: TProc;
+        FProc: TConstFunc<T, TState>;
         function ProcessData(const Value: T): TState;
     public
-        constructor Create(const Controller: IInterface; const AProc: TProc);
+        constructor Create(const Controller: IInterface; const AProc: TConstFunc<T, TState>);
     end;
 
     // Endpoint that collects data into a series.
@@ -216,6 +215,33 @@ type
         property Timeframe: TTimeframe read GetTimeframe;
     end;
 
+    // Endpoint that collects data into a series.
+    TMycDataJoin<T> = class(TInterfacedObject, TDataProvider<TArray<T>>.IDataProvider)
+        type
+            PItem = ^TItem;
+            TItem = record
+                Next: PItem;
+                Value: T;
+            end;
+
+    private
+        FReceivers: array of record
+            DataProvider: TDataProvider<T>;
+            Processor: TMycContainedProcessor<T>;
+            Tag: TDataProvider<T>.TTag;
+            Queue: TQueue<T>;
+        end;
+
+        FSender: TMycContainedDataProvider<TArray<T>>;
+        FLock: TSpinLock;
+        FCount: Integer;
+        function ProcessData(Idx: Integer; const Value: T): TState;
+    public
+        constructor Create(const ADataProviders: TArray<TDataProvider<T>>);
+        destructor Destroy; override;
+        property Sender: TMycContainedDataProvider<TArray<T>> read FSender implements TDataProvider<TArray<T>>.IDataProvider;
+    end;
+
 implementation
 
 uses
@@ -401,12 +427,6 @@ begin
     var TypeT := Context.GetType(TypeInfo(T));
 
     var Fields := Context.GetType(TypeInfo(S)).GetFields;
-    var name := '';
-    if AFieldName = 'Time' then
-        for var i := 0 to High(Fields) do
-        begin
-            name := name + ' ' + Fields[i].Name;
-        end;
 
     Assert(Assigned(Field), 'Field ' + AFieldName + ' not found');
     Assert(Field.FieldType.TypeKind = TypeT.TypeKind, 'Incorrect type');
@@ -444,7 +464,7 @@ end;
 
 { TMycContainedProcessor<T> }
 
-constructor TMycContainedProcessor<T>.Create(const Controller: IInterface; const AProc: TProc);
+constructor TMycContainedProcessor<T>.Create(const Controller: IInterface; const AProc: TConstFunc<T, TState>);
 begin
     inherited Create(Controller);
     FProc := AProc;
@@ -803,4 +823,79 @@ begin
     end;
 end;
 
+{ TMycDataJoin<T> }
+
+constructor TMycDataJoin<T>.Create(const ADataProviders: TArray<TDataProvider<T>>);
+begin
+    inherited Create;
+
+    SetLength(FReceivers, Length(ADataProviders));
+
+    FLock := TSpinLock.Create(false);
+    FCount := Length(FReceivers);
+
+    FSender := TMycContainedDataProvider<TArray<T>>.Create(Self);
+
+    var cFunc :=
+        function(Idx: Integer): TConstFunc<T, TState>
+        begin
+            Result := function(const Value: T): TState begin ProcessData(Idx, Value); end
+        end;
+
+    for var i := 0 to High(FReceivers) do
+        with FReceivers[i] do
+        begin
+            Queue := TQueue<T>.Create;
+            DataProvider := ADataProviders[i];
+            Processor := TMycContainedProcessor<T>.Create(Self, cFunc(i));
+            Tag := DataProvider.Link(Processor);
+        end;
+end;
+
+destructor TMycDataJoin<T>.Destroy;
+begin
+    for var i := High(FReceivers) downto 0 do
+        with FReceivers[i] do
+        begin
+            DataProvider.Unlink(FReceivers[i].Tag);
+            Processor.Free;
+            Queue.Free;
+        end;
+
+    FSender.Free;
+    inherited;
+end;
+
+function TMycDataJoin<T>.ProcessData(Idx: Integer; const Value: T): TState;
+begin
+    var Arr: TArray<T>;
+
+    FLock.Enter;
+    try
+        if FReceivers[Idx].Queue.Count = 0 then
+            dec(FCount);
+
+        if FCount = 0 then
+        begin
+            SetLength(Arr, Length(FReceivers));
+            Arr[Idx] := Value;
+            FCount := Length(FReceivers);
+            for var i := 0 to High(FReceivers) do
+                if i <> Idx then
+                begin
+                    Arr[i] := FReceivers[i].Queue.Dequeue;
+                    if FReceivers[i].Queue.Count > 0 then
+                        dec(FCount);
+                end;
+        end
+        else
+            FReceivers[Idx].Queue.Enqueue(Value);
+    finally
+        FLock.Exit;
+    end;
+
+    if Arr <> nil then
+        FSender.Broadcast(Arr);
+end;
+
 end.