TWriteable<T>

Src/Myc.Trade.DataStream.pas

@@ -54,23 +54,14 @@ type
         function IsHistory: Boolean; virtual; abstract;
     end;
 
-    // Represents a factory for creating IDataStream instances.
-    IDataStreamNode<T> = interface
-        ['{DA3531F1-158E-4A63-8E5A-34089C537B1B}']
-        function CreateDataStream: IDataStream<T>;
-    end;
-
-    // Abstract base class for IDataStreamNode implementations.
-    TDataStreamNode<T> = class(TInterfacedObject, IDataStreamNode<T>)
-    public
-        function CreateDataStream: IDataStream<T>; virtual; abstract;
-    end;
+    TDataProc<T> = reference to procedure(const Values: TArray<TDataPoint<T>>; const Terminated: TState);
 
     // Interface for an instantiable data server.
     IDataServer<T: record> = interface
         ['{1F8E5A9D-E92A-44C1-9F3F-C4B82A6E94B3}']
         function CreateStream(const Symbol: String): IDataStream<T>;
         procedure ClearCache;
+        function ProcessData(const Symbol: String; const Terminate: TSignal; const Proc: TDataProc<T>): TState;
         function EnumerateSymbols: TFuture<TArray<String>>;
     end;
 
@@ -110,6 +101,8 @@ type
         // Used by cache to actually load an uncached file.
         function DoLoad(const FileName: string): TFuture<TArray<TDataPoint<T>>>;
 
+        function LoadFile(currFile: TAuraDataFile; Terminated: TState; Proc: TDataProc<T>): TState;
+
     strict private
         class var
             FLoadGate: TLatch;
@@ -130,6 +123,9 @@ type
         function ParseFileName(const FileName: string): TAuraDataFile; virtual; abstract;
         function EnumerateSymbols: TFuture<TArray<String>>;
         function LoadDataFile(const DataFile: TAuraDataFile): TFuture<TArray<TDataPoint<T>>>;
+
+        function ProcessData(const Symbol: String; const Terminate: TSignal; const Proc: TDataProc<T>): TState;
+
         property Path: String read GetPath;
     end;
 
@@ -180,13 +176,8 @@ type
     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;
+
+    private
         class procedure LoadFile(
             DataServer: TAuraDataServer<T>;
             currFile: TAuraDataFile;
@@ -194,6 +185,14 @@ type
             Done: TLatch;
             Proc: TDataProc
         );
+
+    public
+        class function LoadData(
+            DataServer: TAuraDataServer<T>;
+            const Symbol: String;
+            const Terminate: TSignal;
+            const Proc: TDataProc
+        ): TState;
     end;
 
 implementation
@@ -409,11 +408,44 @@ begin
     Result := FPath;
 end;
 
+function TAuraDataServer<T>.ProcessData(const Symbol: String; const Terminate: TSignal; const Proc: TDataProc<T>): TState;
+begin
+    var capProc := Proc;
+    var terminated := TFlag.CreateObserver(Terminate).State;
+
+    var firstFile := FindFirstFile(Symbol);
+
+    var done := TLatch.CreateLatch(1);
+    Result := done.State;
+
+    TaskManager.CreateTask(firstFile.Done, procedure begin LoadFile(firstFile.Value, terminated, capProc).Subscribe(done); end);
+end;
+
 function TAuraDataServer<T>.LoadDataFile(const DataFile: TAuraDataFile): TFuture<TArray<TDataPoint<T>>>;
 begin
     Result := FCachedFiles.GetOrAdd(DataFile.GetFullFileName);
 end;
 
+function TAuraDataServer<T>.LoadFile(currFile: TAuraDataFile; Terminated: TState; Proc: TDataProc<T>): TState;
+begin
+    if not currFile.IsValid or Terminated.IsSet then
+        exit(TState.Null);
+
+    var done := TLatch.CreateLatch(1);
+    Result := done.State;
+
+    var data := FCachedFiles.GetOrAdd(currFile.GetFullFileName);
+
+    TaskManager.CreateTask(
+        data.Done,
+        procedure
+        begin
+            Proc(data.Value, Terminated);
+            LoadFile(currFile.GetNextFile, Terminated, Proc).Subscribe(done);
+        end
+    );
+end;
+
 { TAuraFileStream<T> }
 
 constructor TAuraFileStream<T>.Create(ADataServer: TAuraDataServer<T>; const ASymbol: String);