MycLib/Src/Myc.Data.Pipeline.pas

unit Myc.Data.Pipeline;

interface

{$M+}

uses
    Myc.Signals,
    Myc.Mutable,
    Myc.Data.Series,
    Myc.Data.Records;

type
    TTag = Pointer;

    // A generic interface for components that consume data of type T.
    IConsumer<T> = interface
        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);
    end;

    // A converter is a producer that internally uses a consumer to transform data.
    IConverter<S, T> = interface(IProducer<T>)
        {$region 'private'}
        function GetConsumer: IConsumer<S>;
        {$endregion}
        property Consumer: IConsumer<S> read GetConsumer;
    end;

    TConvertFunc<S, T> = reference to function(const Value: S): T;
    TBroadcastFunc<T> = reference to function(const Value: T): TState;
    TAggregateFunc<S, T> = reference to function(const Value: S; const Broadcast: TBroadcastFunc<T>): TState;

    // Interface helper for IProducer providing the null object pattern nad subscriptions for linked consumers.
    TProducer<T> = record
    public
        type
            TSubscription = record
            private
                FOwner: IProducer<T>;
                FTag: TTag;
            public
                procedure Unlink;
                class operator Implicit(A: TSubscription): TProducer<T>; overload;
                property Owner: IProducer<T> read FOwner;
            end;

    private
        FProducer: IProducer<T>;
        class function GetNull: IProducer<T>; static; inline;

    public
        constructor Create(const AProducer: IProducer<T>);

        // Managed record operators
        class operator Initialize(out Dest: TProducer<T>);
        class operator Implicit(const A: IProducer<T>): TProducer<T>; overload;
        class operator Implicit(const A: TProducer<T>): IProducer<T>; overload;

        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: TConvertFunc<T, R>): TProducer<R>; overload; inline;
        // Extracts the field of a record by it's name (using RTTI).
        function Field<R>(const FieldName: String): TProducer<R>; inline;

        function MakeParallel: TProducer<T>; inline;

        // Provides access to the null object instance.
        class property Null: IProducer<T> read GetNull;
    end;

    // Interface helper for IConverter<S,T> providing the null object pattern.
    TConverter<S, T> = record
    private
        FConverter: IConverter<S, T>;
        function GetConsumer: IConsumer<S>; inline;
        function GetProducer: TProducer<T>; inline;
        class function GetNull: IConverter<S, T>; static; inline;
    public
        constructor Create(const AConverter: IConverter<S, T>);

        // Managed record operators
        class operator Initialize(out Dest: TConverter<S, T>);
        class operator Implicit(const A: IConverter<S, T>): TConverter<S, T>; overload;
        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 CreateConverter(const Func: TConvertFunc<S, T>): TConverter<S, T>; static;
        class function CreateAggregation(const Func: TAggregateFunc<S, T>): TConverter<S, T>; static;

        property Consumer: IConsumer<S> read GetConsumer;
        property Producer: TProducer<T> read GetProducer;

        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 CreateIdentity<T>: TConverter<T, T>; static;

        class function CreateEndpoint<T>(Lookback: Int64; out Series: TLazy<TSeries<T>>): IConsumer<T>; static;

        type
            TJoinMode = (jmAll, jmAny);

        class function Join<T>(Mode: TJoinMode; const Producers: TArray<TProducer<T>>): TProducer<TArray<T>>; static;

        type
            TRecordMapping = record
                Layout: TDataRecord.TLayout;
                Fields: array of record
                FromField, ToField: String
                end;
            end;

        class function DataMapping(
            const Inputs: TArray<TRecordMapping>;
            const Output: TDataRecord.TLayout
        ): TConverter<TArray<TDataRecord>, TDataRecord>; static;

        class function JoinRecords(
            const TargetLayout: TDataRecord.TLayout;
            const Mapping: TArray<TRecordMapping>;
            const Producers: TArray<TProducer<TDataRecord>>
        ): 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
    Myc.Data.Pipeline.Impl;

{ TProducer<T> }

procedure TProducer<T>.TSubscription.Unlink;
begin
    if FTag <> nil then
    begin
        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;
    if not Assigned(FProducer) then
        FProducer := Null;
end;

function TProducer<T>.Chain(const Next: IConsumer<T>): IConsumer<T>;
begin
    FProducer.Link(Next);
    Result := Next;
end;

function TProducer<T>.Chain<R>(const Next: IConverter<T, R>): TProducer<R>;
begin
    FProducer.Link(Next.Consumer);
    Result := Next;
end;

function TProducer<T>.Chain<R>(const Func: TConvertFunc<T, R>): TProducer<R>;
begin
    Result := Chain<R>(TMycGenericConverter<T, R>.Create(Func));
end;

function TProducer<T>.CreateEndpoint(Lookback: Int64; out Series: TLazy<TSeries<T>>): TSubscription;
begin
    Result := CreateLink(TMycDataEndpoint<T>.CreateDataEndpoint(Lookback, Series));
end;

function TProducer<T>.CreateSequence(Count: Integer; out Seq: TArray<IProducer<T>>): TSubscription;
begin
    Result := CreateLink(TMycSequence<T>.CreateSequence(Count, Seq));
end;

function TProducer<T>.Field<R>(const FieldName: String): TProducer<R>;
begin
    Result := Chain<R>(TMycRecordFieldReader<T, R>.Create(FieldName));
end;

class function TProducer<T>.GetNull: IProducer<T>;
begin
    Result := TMycProducer<T>.Null;
end;

class operator TProducer<T>.Initialize(out Dest: TProducer<T>);
begin
    Dest.FProducer := Null;
end;

class operator TProducer<T>.Implicit(const A: IProducer<T>): TProducer<T>;
begin
    Result.Create(A);
end;

class operator TProducer<T>.Implicit(const A: TProducer<T>): IProducer<T>;
begin
    Result := A.FProducer;
end;

function TProducer<T>.CreateLink(const Consumer: IConsumer<T>): TSubscription;
begin
    Result.FOwner := FProducer;
    Result.FTag := FProducer.Link(Consumer);
end;

function TProducer<T>.MakeParallel: TProducer<T>;
begin
    Result := Chain<T>(TMycParallelConverter<T>.Create as IConverter<T, T>);
end;

constructor TConverter<S, T>.Create(const AConverter: IConverter<S, T>);
begin
    FConverter := AConverter;
    if not Assigned(FConverter) then
        FConverter := Null;
end;

class function TConverter<S, T>.Construct(const Consumer: IConsumer<S>; const Producer: TProducer<T>): TConverter<S, T>;
begin
    Result := TMycComposedConverter<S, T>.Create(Consumer, Producer);
end;

class function TConverter<S, T>.CreateAggregation(const Func: TAggregateFunc<S, T>): TConverter<S, T>;
begin
    Result := TMycGenericAggregator<S, T>.Create(Func);
end;

class function TConverter<S, T>.CreateConverter(const Func: TConvertFunc<S, T>): TConverter<S, T>;
begin
    Result := TMycGenericConverter<S, T>.Create(Func);
end;

function TConverter<S, T>.GetConsumer: IConsumer<S>;
begin
    Result := FConverter.Consumer;
end;

function TConverter<S, T>.GetProducer: TProducer<T>;
begin
    Result := FConverter;
end;

class function TConverter<S, T>.GetNull: IConverter<S, T>;
begin
    Result := TMycConverter<S, T>.Null;
end;

class operator TConverter<S, T>.Initialize(out Dest: TConverter<S, T>);
begin
    Dest.FConverter := Null;
end;

class operator TConverter<S, T>.Implicit(const A: IConverter<S, T>): TConverter<S, T>;
begin
    Result.Create(A);
end;

class operator TConverter<S, T>.Implicit(const A: TConverter<S, T>): IConverter<S, T>;
begin
    Result := A.FConverter;
end;

{ TConverter }

class function TConverter.CreateCounter<T>: TConverter<T, Int64>;
begin
    Result := TMycDataCounter<T>.Create;
end;

class function TConverter.CreateEndpoint<T>(Lookback: Int64; out Series: TLazy<TSeries<T>>): IConsumer<T>;
begin
    Result := TMycDataEndpoint<T>.CreateDataEndpoint(Lookback, Series);
end;

class function TConverter.CreateIdentity<T>: TConverter<T, T>;
begin
    Result := TMycIdentityConverter<T>.Create;
end;

class function TConverter.CreateTicker<T>: TConverter<TArray<T>, T>;
begin
    Result := TMycTicker<T>.Create;
end;

class function TConverter.DataMapping(
    const Inputs: TArray<TRecordMapping>;
    const Output: TDataRecord.TLayout
): TConverter<TArray<TDataRecord>, TDataRecord>;
var
    Idxs: array of array of record
    FromIdx, ToIdx: Integer
    end;
begin
    SetLength(Idxs, Length(Inputs));
    for var i := 0 to High(Idxs) do
    begin
        SetLength(Idxs[i], Length(Inputs[i].Fields));
        for var j := 0 to High(Idxs[i]) do
        begin
            Idxs[i][j].FromIdx := Inputs[i].Layout.IndexOf(Inputs[i].Fields[j].FromField);
            Idxs[i][j].ToIdx := Output.IndexOf(Inputs[i].Fields[j].ToField);
        end;
    end;

    Result :=
        TConverter<TArray<TDataRecord>, TDataRecord>.CreateConverter(
            function(const Inputs: TArray<TDataRecord>): TDataRecord
            begin
                Result := TDataRecord.Create(Output);
                for var i := 0 to High(Idxs) do
                    for var j := 0 to High(Idxs[i]) do
                        Result.CopyValue(Inputs[i], Idxs[i][j].FromIdx, Idxs[i][j].ToIdx);
            end
        );
end;

class function TConverter.Join<T>(Mode: TJoinMode; const Producers: TArray<TProducer<T>>): TProducer<TArray<T>>;
var
    joiner: IDataJoin<T>;
begin
    case Mode of
        jmAll: joiner := TMycDataJoinAll<T>.Create(Length(Producers));
        jmAny: joiner := TMycDataJoinAny<T>.Create(Length(Producers));
    else
        Assert(false);
    end;
    Result := joiner;

    for var i := 0 to High(Producers) do
        Producers[i].Chain(joiner.Consumers[i]);
end;

class function TConverter.JoinRecords(
    const TargetLayout: TDataRecord.TLayout;
    const Mapping: TArray<TRecordMapping>;
    const Producers: TArray<TProducer<TDataRecord>>
): TConverter<TArray<TDataRecord>, TDataRecord>;
begin
    var RecProvider := Join<TDataRecord>(jmAll, Producers);

    Result := DataMapping(Mapping, TargetLayout);

    RecProvider.Chain(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>(jmAll, Self).Chain(map.Consumer);

    Result := map.Producer;
end;

end.