Process.Update for synchronization

Src/Myc.Trade.DataArray.pas

@@ -15,17 +15,19 @@ type
     private
         FChunks: TArray<TChunk>;
         FCount: Int64;
+        FTotalCount: Int64;
 
         function LogicalToPhysicalIndex(LogicalIndex: Int64): Int64; inline;
         function GetItems(Idx: Int64): T; inline;
     public
-        constructor Create(const AChunks: TArray<TChunk>; ACount: Int64);
+        constructor Create(const AChunks: TArray<TChunk>; ACount, ATotalCount: Int64);
         function Add(const Data: T; Lookback: Int64): TMycDataArray<T>; overload;
         function Add(const Data: array of T; First, Count, Lookback: Int64): TMycDataArray<T>; overload;
         class function CreateEmpty: TMycDataArray<T>; static;
         // Helper to create a data array from a raw TArray.
         class function CreateFromArray(const AData: TArray<T>; First, Count: Integer): TMycDataArray<T>; static;
         property Count: Int64 read FCount;
+        property TotalCount: Int64 read FTotalCount;
         property Items[Idx: Int64]: T read GetItems; default;
     end;
 
@@ -33,10 +35,11 @@ implementation
 
 { TMycDataArray<T> }
 
-constructor TMycDataArray<T>.Create(const AChunks: TArray<TChunk>; ACount: Int64);
+constructor TMycDataArray<T>.Create(const AChunks: TArray<TChunk>; ACount, ATotalCount: Int64);
 begin
     FChunks := AChunks;
     FCount := ACount;
+    FTotalCount := ATotalCount;
 end;
 
 class function TMycDataArray<T>.CreateEmpty: TMycDataArray<T>;
@@ -98,7 +101,7 @@ begin
         end;
     end;
 
-    Result := TMycDataArray<T>.Create(newChunks, newCount);
+    Result := TMycDataArray<T>.Create(newChunks, newCount, FTotalCount + Count);
 end;
 
 function TMycDataArray<T>.Add(const Data: T; Lookback: Int64): TMycDataArray<T>;

Src/Myc.Trade.DataPoint.pas

@@ -3,7 +3,8 @@ unit Myc.Trade.DataPoint;
 interface
 
 uses
-    System.TimeSpan;
+    System.TimeSpan,
+    Myc.Core.Notifier;
 
 type
     // A data record for an Ask/Bid price pair.
@@ -31,11 +32,34 @@ type
 
     IMycProcessor<T> = interface
         function ProcessData(const Value: T): Boolean;
+        procedure Update;
     end;
 
     TMycProcessor<T> = class abstract(TInterfacedObject, IMycProcessor<T>)
     protected
         function ProcessData(const Value: T): Boolean; virtual; abstract;
+        procedure Update; virtual;
+    end;
+
+    TTag = Pointer;
+
+    IMycDataProvider<T> = interface
+        function Link(const Receiver: IMycProcessor<T>): TTag;
+        procedure Unlink(Tag: TTag);
+    end;
+
+    TMycDataProvider<T> = class abstract(TContainedObject, IMycDataProvider<T>)
+    private
+        FListeners: TMycNotifyList<IMycProcessor<T>>;
+    public
+        constructor Create(const Controller: IInterface);
+        destructor Destroy; override;
+        // Notifies all linked processors.
+        procedure Notify(const Func: TMycNotifyList<IMycProcessor<T>>.TNotifyProc);
+        // Link a Processor
+        function Link(const Processor: IMycProcessor<T>): TTag;
+        // Unlink a linked strategy
+        procedure Unlink(Tag: TTag);
     end;
 
     TMycGenericProcessor<T> = class(TMycProcessor<T>)
@@ -43,6 +67,7 @@ type
             TProc = reference to function(const Value: T): Boolean;
     private
         FProc: TProc;
+    protected
         function ProcessData(const Value: T): Boolean; override;
     public
         constructor Create(const AProc: TProc);
@@ -54,6 +79,7 @@ type
     private
         FProc: TProc;
         function ProcessData(const Value: T): Boolean;
+        procedure Update;
     public
         constructor Create(const Controller: IInterface; const AProc: TProc);
     end;
@@ -441,4 +467,58 @@ begin
     Result := FProc(Value);
 end;
 
+procedure TMycContainedProcessor<T>.Update;
+begin
+
+end;
+
+{ TMycDataProvider<S, T> }
+
+constructor TMycDataProvider<T>.Create(const Controller: IInterface);
+begin
+    inherited Create(Controller);
+end;
+
+destructor TMycDataProvider<T>.Destroy;
+begin
+    FListeners.Finalize;
+    inherited Destroy;
+end;
+
+procedure TMycDataProvider<T>.Notify(const Func: TMycNotifyList<IMycProcessor<T>>.TNotifyProc);
+begin
+    FListeners.Lock;
+    try
+        FListeners.Notify(Func);
+    finally
+        FListeners.Release;
+    end;
+end;
+
+function TMycDataProvider<T>.Link(const Processor: IMycProcessor<T>): TTag;
+begin
+    // Add the Processor to the notification list
+    FListeners.Lock;
+    try
+        Result := FListeners.Advise(Processor);
+    finally
+        FListeners.Release;
+    end;
+end;
+
+procedure TMycDataProvider<T>.Unlink(Tag: TTag);
+begin
+    FListeners.Lock;
+    try
+        FListeners.Unadvise(Tag);
+    finally
+        FListeners.Release;
+    end;
+end;
+
+procedure TMycProcessor<T>.Update;
+begin
+
+end;
+
 end.

Src/Myc.Trade.DataStream.pas

@@ -437,7 +437,8 @@ begin
             function(const Data: TArray<TDataPoint<T>>): TState
             begin
                 if not Processor.ProcessData(Data) then
-                    exit( TState.Null );
+                    exit(TState.Null);
+                Processor.Update;
 
                 Result := ProcessFile(nextFileInfo, nextFile, cTerminated, Processor);
             end

Src/Myc.Trade.Ticker.pas

@@ -1,56 +0,0 @@
-unit Myc.Trade.Ticker;
-
-interface
-
-uses
-    System.SysUtils,
-    Myc.Signals,
-    Myc.Lazy,
-    Myc.Trade.DataPoint,
-    Myc.Trade.DataStream,
-    Myc.Trade.DataProvider;
-
-type
-    TTickProc<T> = reference to procedure(const Tick: TDataPoint<T>);
-
-    TTicker<T> = class(TInterfacedObject, TSignal.ISubscriber)
-        type
-            TConsumer = record
-                Proc: TTickProc<T>;
-                Lookback: Int64;
-            end;
-    private
-        FStream: IDataStream<T>;
-        FConsumers: TArray<TConsumer>;
-        FProvider: TMutable<TDataSeries<T>>;
-
-        function Notify: Boolean;
-    public
-        constructor Create(const AStream: IDataStream<T>; const AConsumers: TArray<TConsumer>);
-
-    end;
-
-implementation
-
-constructor TTicker<T>.Create(const AStream: IDataStream<T>; const AConsumers: TArray<TConsumer>);
-begin
-    inherited Create;
-    FStream := AStream;
-    FConsumers := AConsumers;
-
-    var ml: Int64 := 0;
-    for var consumer in FConsumers do
-        if consumer.Lookback > ml then
-            ml := consumer.Lookback;
-
-    FProvider := TDataStreamProvider<T>.Create(ml, 100, AStream);
-end;
-
-{ TTicker<T> }
-
-function TTicker<T>.Notify: Boolean;
-begin
-
-end;
-
-end.