Producer pattern in data types

Src/AST/Myc.Ast.DataStream.pas

@@ -0,0 +1,25 @@
+unit Myc.Ast.DataStream;
+
+interface
+
+uses
+    Myc.Data.Pipeline,
+    Myc.Data.Scalar;
+
+type
+    // 1. IStreamProducer (Source)
+    // Provides data. Corresponds to IProducer<T>.
+    IStreamProducer = IProducer<IScalarRecordSeries>;
+
+    // 2. IStreamConsumer (Sink)
+    // Consumes data. Corresponds to IConsumer<T>.
+    IStreamConsumer = IConsumer<IScalarRecordSeries>;
+
+    // 3. IStreamConverter (Transform/Pipe)
+    // Consumes AND provides data.
+    // IMPORTANT: Inherits from IProducer, allowing safe hard-casts to IStreamProducer.
+    IStreamConverter = IConverter<IScalarRecordSeries, IScalarRecordSeries>;
+
+implementation
+
+end.

Src/AST/Myc.Ast.Types.pas

@@ -25,7 +25,11 @@ type
         stSeries,
         stRecord,
         stRecordSeries,
-        stGenericRecord
+        stGenericRecord,
+        // --- Pipeline Types ---
+        stStreamProducer, // Source
+        stStreamConverter, // Pipe (Source + Sink)
+        stStreamConsumer // Sink
     );
 
     TStaticTypeKindHelper = record helper for TStaticTypeKind
@@ -96,6 +100,15 @@ type
             FText: IStaticType;
         class var
             FKeyword: IStaticType;
+
+        // Stream Types
+        class var
+            FStreamProducer: IStaticType;
+        class var
+            FStreamConverter: IStaticType;
+        class var
+            FStreamConsumer: IStaticType;
+
         class constructor Create;
     public
         // Flyweight accessors
@@ -108,6 +121,11 @@ type
         class property Text: IStaticType read FText;
         class property Keyword: IStaticType read FKeyword;
 
+        // Stream Accessors
+        class property StreamProducer: IStaticType read FStreamProducer;
+        class property StreamConverter: IStaticType read FStreamConverter;
+        class property StreamConsumer: IStaticType read FStreamConsumer;
+
         // Factory functions
         class function CreateSeries(const AElementType: IStaticType): IStaticType; static;
         class function CreateMethod(const AParamTypes: TArray<IStaticType>; const AReturnType: IStaticType): IStaticType; static;
@@ -170,6 +188,10 @@ begin
         stRecord: Result := 'Record';
         stRecordSeries: Result := 'RecordSeries';
         stGenericRecord: Result := 'GenericRecord';
+        // New types
+        stStreamProducer: Result := 'StreamProducer';
+        stStreamConverter: Result := 'StreamConverter';
+        stStreamConsumer: Result := 'StreamConsumer';
     else
         Result := 'ErrorType';
     end;
@@ -660,6 +682,11 @@ begin
     FDateTime := TSimpleStaticType.Create(stDateTime);
     FText := TSimpleStaticType.Create(stText);
     FKeyword := TSimpleStaticType.Create(stKeyword);
+
+    // Initialize stream types
+    FStreamProducer := TSimpleStaticType.Create(stStreamProducer);
+    FStreamConverter := TSimpleStaticType.Create(stStreamConverter);
+    FStreamConsumer := TSimpleStaticType.Create(stStreamConsumer);
 end;
 
 class function TTypes.CreateMethod(const AParamTypes: TArray<IStaticType>; const AReturnType: IStaticType): IStaticType;
@@ -740,6 +767,11 @@ begin
     if (Target.Kind = stOrdinal) and (Source.Kind = stKeyword) then
         exit(True);
 
+    // Stream rules:
+    // A Converter IS a Producer.
+    if (Target.Kind = stStreamProducer) and (Source.Kind = stStreamConverter) then
+        exit(True);
+
     Result := False;
 end;
 

Src/Data/Myc.Data.Value.pas

@@ -10,7 +10,8 @@ uses
     Myc.Data.Scalar,
     Myc.Data.Series,
     Myc.Data.Keyword,
-    Myc.Trade.DataFeed;
+    Myc.Trade.DataFeed,
+    Myc.Ast.DataStream;
 
 type
     TDataValueKind = (
@@ -25,7 +26,11 @@ type
         vkMethod,
         vkInterface,
         vkPointer,
-        vkGeneric
+        vkGeneric,
+        // --- Pipeline / Stream Extensions ---
+        vkStreamProducer, // Pure Source (IProducer)
+        vkStreamConverter, // Transformer (IConverter -> IProducer + IConsumer)
+        vkStreamConsumer // Pure Sink (IConsumer)
     );
 
     TDataValue = record
@@ -49,6 +54,8 @@ type
         var
             FKind: TDataValueKind;
             FScalar: TScalar;
+            // Single interface field reused for all interface-based types (incl. Streams)
+            // to keep record size small and cache-friendly.
             FInterface: IInterface;
 
         function GetIsVoid: Boolean; inline;
@@ -57,6 +64,7 @@ type
         class operator Initialize(out Dest: TDataValue);
         class function Void: TDataValue; inline; static;
 
+        // --- Existing Implicits ---
         class operator Implicit(const AValue: Int64): TDataValue; overload; inline;
         class operator Implicit(const AValue: Double): TDataValue; overload; inline;
         class operator Implicit(const AValue: TScalar): TDataValue; overload; inline;
@@ -64,6 +72,12 @@ type
         class operator Implicit(const AValue: String): TDataValue; overload; inline;
         class operator Implicit(const AValue: TDataValue.TFunc): TDataValue; overload; inline;
 
+        // --- Stream Implicits (NEW) ---
+        class operator Implicit(const AValue: IStreamProducer): TDataValue; overload; inline;
+        class operator Implicit(const AValue: IStreamConsumer): TDataValue; overload; inline;
+        // Note: IStreamConverter is covered by IStreamProducer implicit or explicit factory
+
+        // --- Existing Factories ---
         class function FromIntf<T: IInterface>(const AValue: T): TDataValue; static;
         function AsIntf<T: IInterface>: T; inline;
 
@@ -83,6 +97,12 @@ type
         class function FromGenericRecord(const AValue: IKeywordMapping<TDataValue>): TDataValue; static; inline;
         class function FromSeries(const AValue: ISeries): TDataValue; static; inline;
 
+        // --- Stream Factories (NEW) ---
+        class function FromProducer(const AValue: IStreamProducer): TDataValue; static; inline;
+        class function FromConverter(const AValue: IStreamConverter): TDataValue; static; inline;
+        class function FromConsumer(const AValue: IStreamConsumer): TDataValue; static; inline;
+
+        // --- Existing Accessors ---
         function AsScalar: TScalar; inline;
         function AsMethod: TFunc; inline;
         function AsText: String; inline;
@@ -92,6 +112,11 @@ type
         function AsSeries: ISeries; inline;
         function AsObject: TObject; overload; inline;
 
+        // --- Stream Accessors (NEW) ---
+        function AsProducer: IStreamProducer; inline;
+        function AsConverter: IStreamConverter; inline;
+        function AsConsumer: IStreamConsumer; inline;
+
         function ToString: String;
         property IsVoid: Boolean read GetIsVoid;
         property Kind: TDataValueKind read FKind;
@@ -120,6 +145,32 @@ type
         constructor Create(const ASourceSeries: ISeries; const AMapperFunc: TDataValue.TFunc);
     end;
 
+{ TDataValueKindHelper }
+
+function TDataValueKindHelper.ToString: string;
+begin
+    case Self of
+        vkVoid: Result := 'Void';
+        vkScalar: Result := 'Scalar';
+        vkText: Result := 'Text';
+        vkSeries: Result := 'Series';
+        vkRecordSeries: Result := 'RecordSeries';
+        vkScalarRecord: Result := 'ScalarRecord';
+        vkGenericRecord: Result := 'GenericRecord';
+        vkManagedObject: Result := 'Object';
+        vkMethod: Result := 'Method';
+        vkInterface: Result := 'Interface';
+        vkPointer: Result := 'Pointer';
+        vkGeneric: Result := 'Generic';
+        // New types
+        vkStreamProducer: Result := 'StreamProducer';
+        vkStreamConverter: Result := 'StreamConverter';
+        vkStreamConsumer: Result := 'StreamConsumer';
+    else
+        Result := 'Unknown';
+    end;
+end;
+
 { TDataValue.TVal<T> }
 
 constructor TDataValue.TVal<T>.Create(const AValue: T);
@@ -139,6 +190,178 @@ begin
     Dest.FInterface := nil;
 end;
 
+function TDataValue.GetIsVoid: Boolean;
+begin
+    Result := FKind = vkVoid;
+end;
+
+class function TDataValue.Void: TDataValue;
+begin
+    Result.FKind := vkVoid;
+    Result.FInterface := nil;
+end;
+
+// --- Implicit Operators (Existing) ---
+
+class operator TDataValue.Implicit(const AValue: Int64): TDataValue;
+begin
+    Result.FKind := vkScalar;
+    Result.FScalar := TScalar.FromInt64(AValue);
+    Result.FInterface := nil;
+end;
+
+class operator TDataValue.Implicit(const AValue: Double): TDataValue;
+begin
+    Result.FKind := vkScalar;
+    Result.FScalar := TScalar.FromDouble(AValue);
+    Result.FInterface := nil;
+end;
+
+class operator TDataValue.Implicit(const AValue: TScalar): TDataValue;
+begin
+    Result.FKind := vkScalar;
+    Result.FScalar := AValue;
+    Result.FInterface := nil;
+end;
+
+class operator TDataValue.Implicit(const AValue: String): TDataValue;
+begin
+    Result.FKind := vkText;
+    Result.FInterface := TVal<String>.Create(AValue);
+end;
+
+class operator TDataValue.Implicit(const AValue: TDataValue.TFunc): TDataValue;
+begin
+    Result.FKind := vkMethod;
+    TFunc(Result.FInterface) := AValue;
+end;
+
+class operator TDataValue.Implicit(const AValue: TDataValue): TScalar;
+begin
+    if AValue.FKind = vkScalar then
+        Result := AValue.FScalar
+    else
+        raise EInvalidCast.Create('Value is not a scalar');
+end;
+
+// --- Implicit Operators (Streams) ---
+
+class operator TDataValue.Implicit(const AValue: IStreamProducer): TDataValue;
+begin
+    Result := FromProducer(AValue);
+end;
+
+class operator TDataValue.Implicit(const AValue: IStreamConsumer): TDataValue;
+begin
+    Result := FromConsumer(AValue);
+end;
+
+// --- Factories (Streams) ---
+
+class function TDataValue.FromProducer(const AValue: IStreamProducer): TDataValue;
+begin
+    Result.FKind := vkStreamProducer;
+    Result.FInterface := AValue;
+end;
+
+class function TDataValue.FromConverter(const AValue: IStreamConverter): TDataValue;
+begin
+    Result.FKind := vkStreamConverter;
+    Result.FInterface := AValue;
+end;
+
+class function TDataValue.FromConsumer(const AValue: IStreamConsumer): TDataValue;
+begin
+    Result.FKind := vkStreamConsumer;
+    Result.FInterface := AValue;
+end;
+
+// --- Accessors (Streams) ---
+
+function TDataValue.AsProducer: IStreamProducer;
+begin
+    // Safe Hard-Cast: We know FInterface is compatible via FKind.
+    if FKind in [vkStreamProducer, vkStreamConverter] then
+        Result := IStreamProducer(Pointer(FInterface))
+    else
+        raise EInvalidCast.Create('Value is not a StreamProducer');
+end;
+
+function TDataValue.AsConverter: IStreamConverter;
+begin
+    if FKind = vkStreamConverter then
+        Result := IStreamConverter(Pointer(FInterface))
+    else
+        raise EInvalidCast.Create('Value is not a StreamConverter');
+end;
+
+function TDataValue.AsConsumer: IStreamConsumer;
+begin
+    if FKind = vkStreamConsumer then
+        Result := IStreamConsumer(Pointer(FInterface))
+    else
+        raise EInvalidCast.Create('Value is not a StreamConsumer');
+end;
+
+// --- Existing Factories & Accessors ---
+
+class function TDataValue.FromIntf<T>(const AValue: T): TDataValue;
+begin
+    Result.FKind := vkInterface;
+    Result.FInterface := IInterface(AValue);
+end;
+
+class function TDataValue.FromGeneric<T>(const [ref] AValue: T): TDataValue;
+begin
+    Result.FKind := vkGeneric;
+    if PTypeInfo(TypeInfo(T)).Kind = tkInterface then
+        Result.FInterface := IInterface(Pointer(@AValue)^)
+    else
+        Result.FInterface := TVal<T>.Create(AValue);
+end;
+
+class function TDataValue.FromPtr(const AValue: Pointer): TDataValue;
+begin
+    Assert(sizeof(Pointer) <= sizeof(Int64));
+    Result.FKind := vkPointer;
+    Result.FScalar.FromInt64(Int64(AValue));
+end;
+
+procedure TDataValue.FromObj(const AValue: TObject);
+begin
+    FKind := vkManagedObject;
+    FInterface := TObjVal.Create(AValue);
+end;
+
+class function TDataValue.FromSeries(const AValue: ISeries): TDataValue;
+begin
+    Result.FKind := vkSeries;
+    Result.FInterface := AValue;
+end;
+
+class function TDataValue.FromRecordSeries(const AValue: IWriteableScalarRecordSeries): TDataValue;
+begin
+    Result.FKind := vkRecordSeries;
+    Result.FInterface := AValue;
+end;
+
+class function TDataValue.FromScalarRecord(const AValue: IScalarRecord): TDataValue;
+begin
+    Result.FKind := vkScalarRecord;
+    Result.FInterface := AValue;
+end;
+
+class function TDataValue.FromGenericRecord(const AValue: IKeywordMapping<TDataValue>): TDataValue;
+begin
+    Result.FKind := vkGenericRecord;
+    Result.FInterface := AValue;
+end;
+
+class function TDataValue.Map(const SourceSeries: ISeries; const MapperFunc: TFunc): ISeries;
+begin
+    Result := TMapSeries.Create(SourceSeries, MapperFunc);
+end;
+
 function TDataValue.AsSeries: ISeries;
 begin
     if (FKind <> vkSeries) then
@@ -226,75 +449,6 @@ begin
     Result := (FInterface as TVal<String>).Value;
 end;
 
-class function TDataValue.FromIntf<T>(const AValue: T): TDataValue;
-begin
-    Result.FKind := vkInterface;
-    Result.FInterface := IInterface(AValue);
-end;
-
-class function TDataValue.FromGeneric<T>(const [ref] AValue: T): TDataValue;
-begin
-    Result.FKind := vkGeneric;
-    if PTypeInfo(TypeInfo(T)).Kind = tkInterface then
-        Result.FInterface := IInterface(Pointer(@AValue)^)
-    else
-        Result.FInterface := TVal<T>.Create(AValue);
-end;
-
-class function TDataValue.FromPtr(const AValue: Pointer): TDataValue;
-begin
-    Assert(sizeof(Pointer) <= sizeof(Int64));
-    Result.FKind := vkPointer;
-    Result.FScalar.FromInt64(Int64(AValue));
-end;
-
-class function TDataValue.FromSeries(const AValue: ISeries): TDataValue;
-begin
-    Result.FKind := vkSeries;
-    Result.FInterface := AValue;
-end;
-
-class function TDataValue.FromScalarRecord(const AValue: IScalarRecord): TDataValue;
-begin
-    Result.FKind := vkScalarRecord;
-    Result.FInterface := AValue;
-end;
-
-class function TDataValue.FromGenericRecord(const AValue: IKeywordMapping<TDataValue>): TDataValue;
-begin
-    Result.FKind := vkGenericRecord;
-    Result.FInterface := AValue;
-end;
-
-class function TDataValue.FromRecordSeries(const AValue: IWriteableScalarRecordSeries): TDataValue;
-begin
-    Result.FKind := vkRecordSeries;
-    Result.FInterface := AValue;
-end;
-
-class operator TDataValue.Implicit(const AValue: TScalar): TDataValue;
-begin
-    Result.FKind := vkScalar;
-    Result.FScalar := AValue;
-    Result.FInterface := nil;
-end;
-
-class operator TDataValue.Implicit(const AValue: String): TDataValue;
-begin
-    Result.FKind := vkText;
-    Result.FInterface := TVal<String>.Create(AValue);
-end;
-
-function TDataValue.GetIsVoid: Boolean;
-begin
-    Result := FKind = vkVoid;
-end;
-
-class function TDataValue.Map(const SourceSeries: ISeries; const MapperFunc: TFunc): ISeries;
-begin
-    Result := TMapSeries.Create(SourceSeries, MapperFunc);
-end;
-
 function TDataValue.ToString: String;
 var
     sb: TStringBuilder;
@@ -309,7 +463,6 @@ begin
         vkSeries:
         begin
             var series := AsSeries;
-            // Add type and count for series
             Result := Format('<series[%d]>', [series.Count]);
         end;
         vkRecordSeries:
@@ -328,7 +481,6 @@ begin
                     first := False;
                 end;
                 sb.Append('}');
-                // Add field names and count for record series
                 Result := Format('<record_series%s[%d]>', [sb.ToString, series.Count]);
             finally
                 sb.Free;
@@ -349,13 +501,11 @@ begin
                     first := False;
                 end;
                 sb.Append('}');
-                // Add field names for records
                 Result := Format('<scalar_record%s>', [sb.ToString]);
             finally
                 sb.Free;
             end;
         end;
-
         vkGenericRecord:
         begin
             var rec := AsGenericRecord;
@@ -379,73 +529,18 @@ begin
         end;
         vkVoid: Result := '<void>';
         vkMethod: Result := '<method>';
-        vkGeneric:
+        vkGeneric: Result := '<generic>';
-            // Getting meaningful type information for vkGeneric is not easily possible
+        vkManagedObject: Result := 'Object: ' + AsObject.ClassName;
-            // without storing additional RTTI, as the specific type of T in TVal<T> is lost.
+
-            Result := '<generic>';
+        // New Stream Types
+        vkStreamProducer: Result := 'StreamProducer';
+        vkStreamConverter: Result := 'StreamConverter';
+        vkStreamConsumer: Result := 'StreamConsumer';
     else
         Result := '[Unknown DataValue]';
     end;
 end;
 
-class function TDataValue.Void: TDataValue;
-begin
-    Result := Default(TDataValue);
-end;
-
-class operator TDataValue.Implicit(const AValue: TDataValue.TFunc): TDataValue;
-begin
-    Result.FKind := vkMethod;
-    TFunc(Result.FInterface) := AValue;
-end;
-
-procedure TDataValue.FromObj(const AValue: TObject);
-begin
-    FKind := vkManagedObject;
-    FInterface := TObjVal.Create(AValue);
-end;
-
-class operator TDataValue.Implicit(const AValue: TDataValue): TScalar;
-begin
-    Result := AValue.AsScalar;
-end;
-
-class operator TDataValue.Implicit(const AValue: Int64): TDataValue;
-begin
-    Result.FKind := vkScalar;
-    Result.FScalar := TScalar.FromInt64(AValue);
-    Result.FInterface := nil;
-end;
-
-class operator TDataValue.Implicit(const AValue: Double): TDataValue;
-begin
-    Result.FKind := vkScalar;
-    Result.FScalar := TScalar.FromDouble(AValue);
-    Result.FInterface := nil;
-end;
-
-{ TDataValueKindHelper }
-
-function TDataValueKindHelper.ToString: string;
-begin
-    case Self of
-        vkVoid: Result := 'Void';
-        vkScalar: Result := 'Scalar';
-        vkText: Result := 'Text';
-        vkSeries: Result := 'Series';
-        vkRecordSeries: Result := 'RecordSeries';
-        vkScalarRecord: Result := 'ScalarRecord';
-        vkGenericRecord: Result := 'GenericRecord';
-        vkMethod: Result := 'Method';
-        vkInterface: Result := 'Interface';
-        vkPointer: Result := 'Pointer';
-        vkGeneric: Result := 'Generic';
-        vkManagedObject: Result := 'ManagedObject';
-    else
-        Result := 'unknown';
-    end;
-end;
-
 { TMapSeries }
 
 constructor TMapSeries.Create(const ASourceSeries: ISeries; const AMapperFunc: TDataValue.TFunc);
@@ -471,7 +566,6 @@ var
     argArray: TArray<TDataValue>;
     mappedValue: TDataValue;
 begin
-    // Get the original item, apply the mapper function, and return the result.
     sourceValue := FSourceSeries.Items[Idx];
     argArray := [TDataValue(sourceValue)];
     mappedValue := FMapperFunc(argArray);

Src/Myc.Data.Records.pas

@@ -18,7 +18,7 @@ type
                 FFieldType: TFieldType;
                 FTypeInfo: PtypeInfo;
                 FName: string;
-                FOffset: Integer;
+                d FOffset: Integer;
                 FSize: Integer;
 
                 procedure FromType(const [ref] Buffer: TBytes; const Src);