Bugfix in TEvent, DataServer Push-Mode

AuraTrader/AuraTrader.dpr

@@ -5,7 +5,8 @@ uses
   System.StartUpCopy,
   FMX.Forms,
   MainForm in 'MainForm.pas' {Form1},
-  Myc.Trade.Core.DataPoint in '..\Src\Myc.Trade.Core.DataPoint.pas';
+  Myc.Trade.Core.DataPoint in '..\Src\Myc.Trade.Core.DataPoint.pas',
+  Myc.Trade.Ticker in '..\Src\Myc.Trade.Ticker.pas';
 
 {$R *.res}
 

AuraTrader/AuraTrader.dproj

@@ -134,6 +134,7 @@
             <Form>Form1</Form>
         </DCCReference>
         <DCCReference Include="..\Src\Myc.Trade.Core.DataPoint.pas"/>
+        <DCCReference Include="..\Src\Myc.Trade.Ticker.pas"/>
         <BuildConfiguration Include="Base">
             <Key>Base</Key>
         </BuildConfiguration>
@@ -177,9 +178,9 @@
                         <Overwrite>true</Overwrite>
                     </Platform>
                 </DeployFile>
-                <DeployFile LocalName="Win64\Debug\AuraTrader.rsm" Configuration="Debug" Class="DebugSymbols">
+                <DeployFile LocalName="Win64\Release\AuraTrader.exe" Configuration="Release" Class="ProjectOutput">
                     <Platform Name="Win64">
-                        <RemoteName>AuraTrader.rsm</RemoteName>
+                        <RemoteName>AuraTrader.exe</RemoteName>
                         <Overwrite>true</Overwrite>
                     </Platform>
                 </DeployFile>

AuraTrader/MainForm.fmx

@@ -64,6 +64,22 @@ object Form1: TForm1
       JustifyLastLine = Left
       FlowDirection = LeftToRight
     end
+    object Button1: TButton
+      Position.X = 712.00000000000000000
+      Position.Y = 72.00000000000000000
+      TabOrder = 6
+      Text = 'Button1'
+      TextSettings.Trimming = None
+      OnClick = Button1Click
+    end
+    object Button2: TButton
+      Position.X = 720.00000000000000000
+      Position.Y = 96.00000000000000000
+      TabOrder = 8
+      Text = 'Button2'
+      TextSettings.Trimming = None
+      OnClick = Button2Click
+    end
   end
   object LogMemo: TMemo
     Touch.InteractiveGestures = [Pan, LongTap, DoubleTap]

AuraTrader/MainForm.pas

@@ -32,6 +32,7 @@ uses
     Myc.Signals,
     Myc.Lazy,
     Myc.Signals.FMX,
+    Myc.TaskManager,
     FMX.ListBox,
     FMX.Layouts;
 
@@ -44,10 +45,14 @@ type
         SymbolsComboBox: TComboBox;
         LoadButton: TButton;
         FlowLayout: TFlowLayout;
+        Button1: TButton;
+        Button2: TButton;
         procedure RandomButtonClick(Sender: TObject);
         procedure FormCreate(Sender: TObject);
         procedure FormDestroy(Sender: TObject);
         procedure LoadButtonClick(Sender: TObject);
+        procedure Button1Click(Sender: TObject);
+        procedure Button2Click(Sender: TObject);
     private
         const
             cnt = 20;
@@ -63,6 +68,8 @@ type
         FServer: IDataServer<TAskBidItem>;
         FSymbols: TFuture<TArray<String>>;
         FRandom: TList<TRndItem>;
+        FTerminate: TEvent;
+        FLoadDone: TState;
     public
     { Public declarations }
     published
@@ -76,8 +83,61 @@ implementation
 
 {$R *.fmx}
 
+procedure TForm1.Button1Click(Sender: TObject);
+begin
+    if SymbolsComboBox.ItemIndex < 0 then
+        exit;
+
+    var arr := TDataSeries<TAskBidItem>.CreateArray(3000);
+    var curr := TMutable<TDataSeries<TAskBidItem>>.CreateProtected;
+    var Symbol := FSymbols.WaitFor[SymbolsComboBox.ItemIndex];
+
+    var done :=
+        TAuraFileLoader<TAskBidItem>.LoadData(
+            FServer as TAuraTABFileServer,
+            Symbol,
+            FTerminate.Signal,
+            procedure(const Values: TArray<TDataPoint<TAskBidItem>>)
+            begin
+                arr.Add(Values);
+                curr.Value := arr.Copy;
+            end
+        );
+
+    FLoadDone := TState.All([FLoadDone, done]);
+
+    LogMemo.AddIdleHandler(curr.Changed, procedure begin LogMemo.Lines.Add(Symbol + ': ' + curr.Value.TotalCount.ToString); end);
+
+    Path1.AddIdleHandler(
+        curr.Changed,
+        procedure
+        begin
+            var Prices := curr.Value;
+
+            if Prices.Count > 0 then
+            begin
+                Path1.BeginUpdate;
+                try
+                    Path1.Data.Clear;
+                    Path1.Data.MoveTo(PointF(Path1.Width - 1, Prices[0].Data.Ask));
+                    for var i := 1 to Prices.Count - 1 do
+                        Path1.Data.LineTo(PointF(Path1.Width - i - 1, Prices[i].Data.Ask));
+                finally
+                    Path1.EndUpdate;
+                end;
+            end;
+        end
+    );
+end;
+
+procedure TForm1.Button2Click(Sender: TObject);
+begin
+    FTerminate.Notify;
+end;
+
 procedure TForm1.FormCreate(Sender: TObject);
 begin
+    FTerminate := TEvent.CreateEvent;
     FRandom := TList<TRndItem>.Create;
 
     // Create an instance of the TAuraTABFileServer. The server can be reused for multiple stream creations. [364]
@@ -110,6 +170,8 @@ end;
 
 procedure TForm1.FormDestroy(Sender: TObject);
 begin
+    FTerminate.Notify;
+    TaskManager.WaitFor(FLoadDone);
     Application.OnIdle := nil;
     FRandom.Free;
 end;

Src/Myc.Core.Lazy.pas

@@ -3,6 +3,7 @@ unit Myc.Core.Lazy;
 interface
 
 uses
+    System.Classes,
     System.SysUtils,
     Myc.Signals,
     Myc.Lazy;
@@ -76,6 +77,19 @@ type
         constructor Create(const AChanged: TSignal.ISignal; const AProc: TFunc<T>);
     end;
 
+    TMycProtectedWriteableMutable<T> = class(TInterfacedObject, TMutable<T>.IMutable, TMutable<T>.IWriteable)
+    private
+        FValue: T;
+        FChanged: TEvent;
+        FLock: Integer;
+    protected
+        function GetChanged: TSignal;
+        function GetValue: T;
+    public
+        constructor Create(const AValue: T);
+        procedure SetValue(const Value: T);
+    end;
+
 implementation
 
 { TMycNullMutable<T> }
@@ -204,4 +218,41 @@ begin
     Result := FProc();
 end;
 
+{ TMycProtectedWriteableMutable<T> }
+
+constructor TMycProtectedWriteableMutable<T>.Create(const AValue: T);
+begin
+    inherited Create;
+    FValue := AValue;
+    FChanged := TEvent.CreateEvent;
+end;
+
+function TMycProtectedWriteableMutable<T>.GetChanged: TSignal;
+begin
+    Result := FChanged.Signal;
+end;
+
+function TMycProtectedWriteableMutable<T>.GetValue: T;
+begin
+    while AtomicExchange(FLock, 1) = 1 do
+        YieldProcessor;
+    try
+        Result := FValue;
+    finally
+        AtomicExchange(FLock, 0);
+    end;
+end;
+
+procedure TMycProtectedWriteableMutable<T>.SetValue(const Value: T);
+begin
+    while AtomicExchange(FLock, 1) = 1 do
+        YieldProcessor;
+    try
+        FValue := Value;
+        FChanged.Notify;
+    finally
+        AtomicExchange(FLock, 0);
+    end;
+end;
+
 end.

Src/Myc.Core.Signals.pas

@@ -217,7 +217,6 @@ begin
 
     FSubscribers.Lock;
     try
-        Subscriber.Notify;
         Result := FSubscribers.Advise(Subscriber);
     finally
         FSubscribers.Release;

Src/Myc.Core.Tasks.pas

@@ -82,7 +82,7 @@ type
         procedure EnqueueJob(const Job: TProc);
         function Run(Job: TProc): TSignal.ISubscriber;
         function CreateTask(const Gate: TState; const Proc: TProc): TSignal.TSubscription;
-        procedure WaitFor(State: TState.IState);
+        procedure WaitFor(const State: TState.IState);
         procedure Teardown;
         function InMainThread: Boolean;
         function InWorkerThread: Boolean;
@@ -345,7 +345,7 @@ begin
     end;
 end;
 
-procedure TMycTaskFactory.WaitFor(State: TState.IState);
+procedure TMycTaskFactory.WaitFor(const State: TState.IState);
 var
     lock: TSemaphore; // This is System.SyncObjs.TSemaphore
 begin

Src/Myc.Lazy.pas

@@ -20,6 +20,7 @@ type
 
             IWriteable = interface(IMutable)
                 procedure SetValue(const Value: T);
+                property Value: T read GetValue write SetValue;
             end;
 
     {$REGION 'private'}
@@ -44,6 +45,7 @@ type
         class function Construct(const Changing: TSignal; const Proc: TFunc<T>): TMutable<T>; overload; static;
 
         class function CreateWriteable(const Init: T): IWriteable; overload; static;
+        class function CreateProtected: IWriteable; overload; static;
 
         property Value: T read GetValue;
         property Changed: TSignal read GetChanged;
@@ -107,6 +109,11 @@ begin
     Result := TMycFuncMutable<T>.Create(Changing, Proc);
 end;
 
+class function TMutable<T>.CreateProtected: IWriteable;
+begin
+    Result := TMycProtectedWriteableMutable<T>.Create(Default(T));
+end;
+
 class function TMutable<T>.CreateWriteable(const Init: T): IWriteable;
 begin
     Result := TMycWriteableMutable<T>.Create(Init);

Src/Myc.Signals.FMX.pas

@@ -9,7 +9,7 @@ uses
     Myc.Signals;
 
 type
-    TMsgProc = reference to procedure( const Sender: TObject; const M: TMessage );
+    TMsgProc = reference to procedure(const Sender: TObject; const M: TMessage);
 
     TComponentValidation = class(TComponent)
     private
@@ -32,11 +32,12 @@ type
 implementation
 
 uses
-    {$IFDEF FRAMEWORK_FMX}
+{$IFDEF FRAMEWORK_FMX}
     FMX.Types
-    {$ELSE}
+{$ELSE}
     VCL.Types // to be checked
-    {$IFEND};
+{$IFEND}
+        ;
 
 { TComponentValidation }
 
@@ -58,17 +59,13 @@ end;
 procedure TComponentValidation.DoMsg(const Sender: TObject; const M: TMessage);
 begin
     if FReceived.Reset then
-        FProc( Sender, M );
+        FProc(Sender, M);
 end;
 
 procedure TComponentValidationHelper.AddIdleHandler(const Signal: TSignal; const Proc: TProc);
 begin
     var capProc := Proc;
-    AddMsgHandler(TIdleMessage, Signal,
-        procedure( const Sender: TObject; const M: TMessage )
-        begin
-            capProc();
-        end );
+    AddMsgHandler(TIdleMessage, Signal, procedure(const Sender: TObject; const M: TMessage) begin capProc(); end);
 end;
 
 procedure TComponentValidationHelper.AddMsgHandler(const MsgClass: TClass; const Signal: TSignal; const Proc: TMsgProc);

Src/Myc.TaskManager.pas

@@ -16,7 +16,7 @@ type
         // After waiting, or if the state is already set, any first stored exception
         // from any worker thread of this factory will be re-raised in the calling (main) thread.
         // Raises ETaskException if called from within a worker thread.
-        procedure WaitFor(State: TState.IState);
+        procedure WaitFor(const State: TState.IState);
     end;
 
 var
@@ -42,7 +42,7 @@ type
     public
         // IMycTaskManager
         function CreateTask(const Gate: TState; const Proc: TProc): TSignal.TSubscription;
-        procedure WaitFor(State: TState.IState);
+        procedure WaitFor(const State: TState.IState);
 
         constructor Create;
     end;
@@ -77,7 +77,7 @@ begin
     Result := Gate.Signal.Subscribe(TMycExecMock.Create(Proc));
 end;
 
-procedure TMycTaskManagerMock.WaitFor(State: TState.IState);
+procedure TMycTaskManagerMock.WaitFor(const State: TState.IState);
 begin
     Assert(State.IsSet);
 end;

Src/Myc.Trade.Core.DataPoint.pas

@@ -14,7 +14,6 @@ type
             TChunk = TArray<TDataPoint<T>>;
     private
         FChunks: TArray<TChunk>;
-        FChunks1: TArray<TChunk>;
         FCount: Int64;
         FMaxLookback: Int64;
         FTotalCount: Int64;

Src/Myc.Trade.DataStream.pas

@@ -176,6 +176,26 @@ type
         function ParseFileName(const FileName: string): TAuraDataFile; override;
     end;
 
+    // Implements a data stream that reads from Aura-specific historical data files.
+    TAuraFileLoader<T: record> = class(TInterfacedObject)
+        type
+            TDataProc = reference to procedure(const Values: TArray<TDataPoint<T>>);
+    public
+        class function LoadData(
+            DataServer: TAuraDataServer<T>;
+            const Symbol: String;
+            const Terminate: TSignal;
+            const Proc: TDataProc
+        ): TState;
+        class procedure LoadFile(
+            DataServer: TAuraDataServer<T>;
+            currFile: TAuraDataFile;
+            Terminated: TState;
+            Done: TLatch;
+            Proc: TDataProc
+        );
+    end;
+
 implementation
 
 uses
@@ -740,4 +760,49 @@ begin
     end;
 end;
 
+class function TAuraFileLoader<T>.LoadData(
+    DataServer: TAuraDataServer<T>;
+    const Symbol: String;
+    const Terminate: TSignal;
+    const Proc: TDataProc
+): TState;
+begin
+    var capProc := Proc;
+    var terminated := TFlag.CreateObserver(Terminate).State;
+
+    var firstFile := DataServer.FindFirstFile(Symbol);
+
+    var done := TLatch.CreateLatch(1);
+
+    TaskManager.CreateTask(firstFile.Done, procedure begin LoadFile(DataServer, firstFile.Value, terminated, done, capProc); end);
+
+    Result := done.State;
+end;
+
+class procedure TAuraFileLoader<T>.LoadFile(
+    DataServer: TAuraDataServer<T>;
+    currFile: TAuraDataFile;
+    Terminated: TState;
+    Done: TLatch;
+    Proc: TDataProc
+);
+begin
+    if not currFile.IsValid or Terminated.IsSet then
+    begin
+        Done.Notify;
+        exit;
+    end;
+
+    var data := DataServer.LoadDataFile(currFile);
+
+    TaskManager.CreateTask(
+        data.Done,
+        procedure
+        begin
+            Proc(data.Value);
+            LoadFile(DataServer, currFile.GetNextFile, Terminated, Done, Proc);
+        end
+    );
+end;
+
 end.

Src/Myc.Trade.Ticker.pas

@@ -3,70 +3,54 @@ unit Myc.Trade.Ticker;
 interface
 
 uses
-    Myc.Lazy;
+    System.SysUtils,
+    Myc.Signals,
+    Myc.Lazy,
+    Myc.Trade.DataPoint,
+    Myc.Trade.DataStream,
+    Myc.Trade.DataProvider;
 
 type
-    TMycLogger = reference to procedure(const Msg: String);
+    TTickProc<T> = reference to procedure(const Tick: TDataPoint<T>);
 
-    IMycTradeObject = interface
-    end;
-
-    TMycTradeObject = class(TInterfacedObject, IMycTradeObject)
+    TTicker<T> = class(TInterfacedObject, TSignal.ISubscriber)
+        type
+            TConsumer = record
+                Proc: TTickProc<T>;
+                Lookback: Int64;
+            end;
     private
-        FCaption: string;
-        FLog: TMycLogger;
-        function GetCaption: string;
+        FStream: IDataStream<T>;
+        FConsumers: TArray<TConsumer>;
+        FProvider: TMutable<TDataSeries<T>>;
+
+        function Notify: Boolean;
     public
-        constructor Create(const ACaption: string; ALog: TMycLogger);
-        property Caption: string read GetCaption;
-        property Log: TMycLogger read FLog;
-    end;
-
-    IMycTime = interface
-        {$REGION 'property access'}
-        function GetTimeStamp: TDateTime;
-        {$ENDREGION}
-        property TimeStamp: TDateTime read GetTimeStamp;
-    end;
-
-    IMycTick = interface
-        {$REGION 'property access'}
-        function GetAsk: Double;
-        function GetBid: Double;
-        function GetTime: IMycTime;
-        function GetVolume: Double;
-        {$ENDREGION}
-        property Ask: Double read GetAsk;
-        property Bid: Double read GetBid;
-        property Time: IMycTime read GetTime;
-        property Volume: Double read GetVolume;
-    end;
-
-    IMycTicker = interface(IMycTradeObject)
-        {$REGION 'property access'}
-        function GetLastTick: TLazy<IMycTick>;
-        function GetCurrentTime: IMycTime;
-        {$ENDREGION}
-        property LastTick: TLazy<IMycTick> read GetLastTick;
-        property CurrentTime: IMycTime read GetCurrentTime;
-    end;
-
-    IMycHistoryTicker = interface(IMycTicker)
+        constructor Create(const AStream: IDataStream<T>; const AConsumers: TArray<TConsumer>);
 
     end;
 
 implementation
 
-constructor TMycTradeObject.Create(const ACaption: string; ALog: TMycLogger);
+constructor TTicker<T>.Create(const AStream: IDataStream<T>; const AConsumers: TArray<TConsumer>);
 begin
     inherited Create;
-    FCaption := ACaption;
-    FLog := procedure(const Msg: String) begin ALog('[' + FCaption + '] ' + Msg); end;
+    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;
 
-function TMycTradeObject.GetCaption: string;
+{ TTicker<T> }
+
+function TTicker<T>.Notify: Boolean;
 begin
-    Result := FCaption;
+
 end;
 
 end.