DataPoint

Src/Myc.Trade.DataStream.pas

@@ -145,6 +145,7 @@ type
         FDataServer: TAuraDataServer<T>;
         FSymbol: String;
         FHasData: TEvent;
+        FNextReady: TFlag;
         FCurrent: TFuture<TDataStream>;
         FNext: TFuture<TDataStream>;
         FCurrPosInFile: Int64;
@@ -166,13 +167,13 @@ type
 
     TAuraTABFileServer = class(TAuraDataServer<TAskBidItem>)
     protected
-        function ParseFileName(const FileName: string): TAuraDataFile; override;
         // Scans a directory and returns the oldest file found for each symbol.
         class function ReadCompressedData(const InputStream: TStream): TArray<TDataPoint<TAskBidItem>>; override;
         class function ReadUncompressedData(const InputStream: TStream): TArray<TDataPoint<TAskBidItem>>; override;
     public
         function CreateStream(const Symbol: String): IDataStream<TAskBidItem>; override;
         function LoadDataSeries(const InitialFile: TAuraDataFile): TFuture<TArray<TDataPoint<TAskBidItem>>>;
+        function ParseFileName(const FileName: string): TAuraDataFile; override;
     end;
 
 implementation
@@ -363,9 +364,6 @@ begin
 end;
 
 function TAuraDataServer<T>.EnumerateSymbols: TFuture<TArray<String>>;
-var
-    dataFiles: TArray<TAuraDataFile>;
-    i: Integer;
 begin
     Result :=
         FSymbols.Chain<TArray<String>>(
@@ -404,7 +402,8 @@ begin
     Assert(Assigned(ADataServer));
     FDataServer := ADataServer;
     FSymbol := ASymbol;
-    FHasData := TEvent.CreateEvent;
+    FNextReady := TFlag.CreateFlag;
+    FHasData := TEvent.CreateRouter(FNextReady.State.AsSignal);
 end;
 
 destructor TAuraFileStream<T>.Destroy;
@@ -429,7 +428,7 @@ begin
                 begin
                     Result.FileInfo := FileInfo;
                     Result.Data := FDataServer.LoadDataFile(FileInfo);
-                    Result.Data.Done.Subscribe(FHasData);
+                    Result.Data.Done.Subscribe(FNextReady);
                 end;
             end
         );
@@ -488,7 +487,7 @@ begin
         Inc(FCurrPosInFile);
     end;
 
-    if FCurrPosInFile < Length(currData) then
+    if (FCurrPosInFile < Length(currData)) or FNextReady.Reset then
     begin
         FHasData.Notify;
     end;
@@ -511,6 +510,8 @@ end;
 
 procedure TAuraFileStream<T>.PreloadNext;
 begin
+    FNextReady.Reset;
+
     FNext :=
         FCurrent.Chain<TDataStream>(
             function(const Prev: TDataStream): TDataStream
@@ -524,7 +525,7 @@ begin
                     begin
                         Result.FileInfo := nextFileInfo;
                         Result.Data := FDataServer.LoadDataFile(nextFileInfo);
-                        Result.Data.Done.Subscribe(FHasData);
+                        Result.Data.Done.Subscribe(FNextReady);
                     end;
                 end;
             end