Data Types

Src/Myc.Data.Pipeline.Impl.pas

@@ -193,8 +193,12 @@ type
         function Consume(const Value: T): TState; override; final;
     end;
 
-    // Endpoint that collects data into a series.
-    TMycDataJoin<T> = class(TInterfacedObject, IProducer<TArray<T>>)
+    IDataJoin<T> = interface(IProducer<TArray<T>>)
+        function GetConsumers(Idx: Integer): IConsumer<T>;
+        property Consumers[Idx: Integer]: IConsumer<T> read GetConsumers;
+    end;
+
+    TMycDataJoinAll<T> = class(TInterfacedObject, IProducer<TArray<T>>, IDataJoin<T>)
     private
         FConsumers: array of record
             Consumer: TMycGenericConsumer<T>;
@@ -214,6 +218,25 @@ type
         property Consumers[Idx: Integer]: IConsumer<T> read GetConsumers;
     end;
 
+    TMycDataJoinAny<T> = class(TInterfacedObject, IProducer<TArray<T>>, IDataJoin<T>)
+    private
+        FConsumers: array of TMycGenericConsumer<T>;
+        FValues: TArray<T>;
+        FInit: TArray<Boolean>;
+        FInitCount: Integer;
+
+        FContainedProvider: TMycContainedProducer<TArray<T>>;
+        FLock: TSpinLock;
+        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(ACount: Integer);
+        destructor Destroy; override;
+        property Consumers[Idx: Integer]: IConsumer<T> read GetConsumers;
+    end;
+
     TMycComposedConverter<S, T> = class(TInterfacedObject, IConverter<S, T>)
     private
         FConsumer: IConsumer<S>;
@@ -589,9 +612,9 @@ begin
     FQueue := Result;
 end;
 
-{ TMycDataJoin<T> }
+{ TMycDataJoinAll<T> }
 
-constructor TMycDataJoin<T>.Create(ACount: Integer);
+constructor TMycDataJoinAll<T>.Create(ACount: Integer);
 begin
     inherited Create;
 
@@ -616,7 +639,7 @@ begin
         end;
 end;
 
-destructor TMycDataJoin<T>.Destroy;
+destructor TMycDataJoinAll<T>.Destroy;
 begin
     for var i := High(FConsumers) downto 0 do
         with FConsumers[i] do
@@ -629,17 +652,17 @@ begin
     inherited;
 end;
 
-function TMycDataJoin<T>.Link(const Consumer: IConsumer<TArray<T>>): TTag;
+function TMycDataJoinAll<T>.Link(const Consumer: IConsumer<TArray<T>>): TTag;
 begin
     Result := FContainedProvider.Link(Consumer);
 end;
 
-procedure TMycDataJoin<T>.Unlink(Tag: TTag);
+procedure TMycDataJoinAll<T>.Unlink(Tag: TTag);
 begin
     FContainedProvider.Unlink(Tag);
 end;
 
-function TMycDataJoin<T>.Consume(Idx: Integer; const Value: T): TState;
+function TMycDataJoinAll<T>.Consume(Idx: Integer; const Value: T): TState;
 begin
     var Arr: TArray<T>;
 
@@ -671,7 +694,7 @@ begin
         FContainedProvider.Broadcast(Arr);
 end;
 
-function TMycDataJoin<T>.GetConsumers(Idx: Integer): IConsumer<T>;
+function TMycDataJoinAll<T>.GetConsumers(Idx: Integer): IConsumer<T>;
 begin
     Result := FConsumers[Idx].Consumer;
 end;
@@ -797,4 +820,75 @@ begin
 
 end;
 
+{ TMycDataJoinAny<T> }
+
+constructor TMycDataJoinAny<T>.Create(ACount: Integer);
+begin
+    inherited Create;
+
+    SetLength(FConsumers, ACount);
+    SetLength(FValues, ACount);
+    SetLength(FInit, ACount);
+    FInitCount := ACount;
+
+    FLock := TSpinLock.Create(false);
+
+    FContainedProvider := TMycContainedProducer<TArray<T>>.Create(Self);
+
+    var cFunc :=
+        function(Idx: Integer): TConvertFunc<T, TState>
+        begin
+            Result := function(const Value: T): TState begin Result := Consume(Idx, Value); end
+        end;
+
+    for var i := 0 to High(FConsumers) do
+    begin
+        FValues[i] := Default(T);
+        FInit[i] := false;
+        FConsumers[i] := TMycGenericConsumer<T>.Create(Self, cFunc(i));
+    end;
+end;
+
+destructor TMycDataJoinAny<T>.Destroy;
+begin
+    for var i := High(FConsumers) downto 0 do
+        FConsumers[i].Free;
+
+    FContainedProvider.Free;
+    inherited;
+end;
+
+function TMycDataJoinAny<T>.Link(const Consumer: IConsumer<TArray<T>>): TTag;
+begin
+    Result := FContainedProvider.Link(Consumer);
+end;
+
+procedure TMycDataJoinAny<T>.Unlink(Tag: TTag);
+begin
+    FContainedProvider.Unlink(Tag);
+end;
+
+function TMycDataJoinAny<T>.Consume(Idx: Integer; const Value: T): TState;
+begin
+    FLock.Enter;
+    try
+        FValues[Idx] := Value;
+        if not FInit[Idx] then
+        begin
+            FInit[Idx] := true;
+            dec(FInitCount);
+        end;
+
+        if FInitCount = 0 then
+            FContainedProvider.Broadcast(FValues);
+    finally
+        FLock.Exit;
+    end;
+end;
+
+function TMycDataJoinAny<T>.GetConsumers(Idx: Integer): IConsumer<T>;
+begin
+    Result := FConsumers[Idx];
+end;
+
 end.