cTrader-DataExport/ChartDisplay/DataStreamer.pas

unit DataStreamer;

interface

uses
    System.SysUtils, System.Classes, System.Generics.Collections, System.IOUtils, System.Threading,
    System.Zip;

type
    IDataPage = interface
        ['{B3E3C5B7-7C1A-4A8B-8B7E-15C9D9E0F4A6}']
        function GetStream: TStream;
        function GetSourceIdentifier: string;
        function GetLastModificationTime: TDateTime;
    end;

    IRecordStreamReader<T: record> = interface
        ['{C1B7E8F0-5A0A-4F3B-BD28-18D5E5A3F7A0}']
        function GetCurrentRecord: T;
        function MoveToNextRecord: Boolean;
        procedure Reset;
        function GetCurrentRecordIndex: Int64;
        function Seek(RecordIndex: Int64): Int64;

        property CurrentRecord: T read GetCurrentRecord;
        property CurrentRecordIndex: Int64 read GetCurrentRecordIndex;
    end;

    TLocalFilePage = class(TInterfacedObject, IDataPage)
    private
        FActualFilePath: string;
        FMemoryStream: TMemoryStream;
        FLastModifiedOnDisk: TDateTime;

        function GetStream: TStream;
        function GetSourceIdentifier: string;
        function GetLastModificationTime: TDateTime;
    public
        constructor Create(const AActualFilePath: string; ADataStream: TMemoryStream; ALastModifiedTimeOnDisk: TDateTime);
        destructor Destroy; override;
        class function IsFileModified(const AFileName: string; KnownLastModified: TDateTime): Boolean;
    end;

    TStreamingFileRecordReader<T: record> = class(TInterfacedObject, IRecordStreamReader<T>)
    private
        FDirectoryPath: string;
        FDataSetName: string;
        FExtension: string;

        FFilePaths: TStringList;
        FCachedPages: TObjectDictionary<string, IDataPage>;
        FPageCacheQueue: TQueue<string>;
        FMaxCacheSize: Integer;
        FPreloadPageTask: ITask;
        FLockObject: TObject;

        FCurrentPage: IDataPage;
        FCurrentPageStream: TStream;
        FCurrentPageIndexInFileList: Integer;

        FRecordSize: Integer;
        FCurrentRecord: T;
        FRecordsInCurrentPage: Int64;
        FRecordsReadFromCurrentPage: Int64;
        FGlobalRecordIndex: Int64;

        procedure ScanDirectoryAndMatchFiles;
        function LoadPage(ActualFilePath: string): IDataPage;
        procedure EnsurePageInCache(ActualFilePath: string; IsPreload: Boolean = False);
        procedure EvictOldestPage;
        procedure DoPreloadWork(const APathToPreload: string);
        procedure StartPreloadNextPage;
        procedure InvalidatePageIfModified(PageToCheck: IDataPage; IsCurrentPage: Boolean);
        function ActivatePage(PageIndexInFileList: Integer): Boolean;
        function GetRecordCountForFile(ActualFilePath: string): Int64;

        function GetCurrentRecord: T;
        function MoveToNextRecord: Boolean;
        procedure Reset;
        function GetCurrentRecordIndex: Int64;
        function Seek(RecordIndex: Int64): Int64;
    public
        constructor Create(const ADirectoryPath: string; const ADataSetName: string; const AExtension: string; AMaxCacheSize: Integer = 2);
        destructor Destroy; override;
    end;

implementation

uses
    System.DateUtils, System.Types, System.StrUtils,
    System.Math;

{ TLocalFilePage }

constructor TLocalFilePage.Create(const AActualFilePath: string; ADataStream: TMemoryStream; ALastModifiedTimeOnDisk: TDateTime);
begin
    inherited Create;
    FActualFilePath := AActualFilePath;
    FMemoryStream := ADataStream;
    FMemoryStream.Position := 0;
    FLastModifiedOnDisk := ALastModifiedTimeOnDisk;
end;

destructor TLocalFilePage.Destroy;
begin
    FMemoryStream.Free;
    inherited Destroy;
end;

function TLocalFilePage.GetStream: TStream;
begin
    FMemoryStream.Position := 0;
    result := FMemoryStream;
end;

function TLocalFilePage.GetSourceIdentifier: string;
begin
    result := FActualFilePath;
end;

function TLocalFilePage.GetLastModificationTime: TDateTime;
begin
    result := FLastModifiedOnDisk;
end;

class function TLocalFilePage.IsFileModified(const AFileName: string; KnownLastModified: TDateTime): Boolean;
var
    currentModTime: TDateTime;
begin
    if TFile.Exists(AFileName) then
    begin
        currentModTime := TFile.GetLastWriteTime(AFileName);
        result := Abs(MilliSecondsBetween(currentModTime, KnownLastModified)) > 1000;
    end
    else
    begin
        result := True;
    end;
end;

{ TStreamingFileRecordReader<T> }

constructor TStreamingFileRecordReader<T>.Create(const ADirectoryPath: string; const ADataSetName: string; const AExtension: string; AMaxCacheSize: Integer);
begin
    inherited Create;
    FLockObject := TObject.Create;

    if not TDirectory.Exists(ADirectoryPath) then
        raise EDirectoryNotFoundException.Create('Directory not found: ' + ADirectoryPath);
    if Trim(ADataSetName) = '' then
        raise EArgumentException.Create('DataSetName cannot be empty.');
    if Trim(AExtension) = '' then
        raise EArgumentException.Create('Extension cannot be empty.');

    FDirectoryPath := IncludeTrailingPathDelimiter(ADirectoryPath);
    FDataSetName := ADataSetName;
    FExtension := AExtension;
    if StartsText('.', FExtension) then
       FExtension := Copy(FExtension, 2, Length(FExtension) -1);

    FFilePaths := TStringList.Create;
    FFilePaths.Sorted := True;

    FCachedPages := TObjectDictionary<string, IDataPage>.Create([doOwnsValues]);
    FPageCacheQueue := TQueue<string>.Create;
    FMaxCacheSize := System.Math.Max(2, AMaxCacheSize);
    FRecordSize := SizeOf(T);
    if FRecordSize = 0 then
        raise EArgumentException.Create('Record size for type T cannot be zero.');

    FCurrentPageIndexInFileList := -1;
    FGlobalRecordIndex := -1;
    FRecordsReadFromCurrentPage := 0;
    FRecordsInCurrentPage := 0;
    FCurrentPage := nil;
    FCurrentPageStream := nil;

    ScanDirectoryAndMatchFiles;
    Reset;
end;

destructor TStreamingFileRecordReader<T>.Destroy;
var
    taskToWaitFor: ITask;
begin
    taskToWaitFor := nil;
    TMonitor.Enter(FLockObject);
    try
        taskToWaitFor := FPreloadPageTask;
        FPreloadPageTask := nil;
    finally
        TMonitor.Exit(FLockObject);
    end;

    if (taskToWaitFor <> nil) and (taskToWaitFor.Status in [TTaskStatus.Created, TTaskStatus.WaitingToRun, TTaskStatus.Running]) then
    begin
        // Consider taskToWaitFor.Cancel; taskToWaitFor.Wait;
    end;

    FFilePaths.Free;
    FCachedPages.Free;
    FPageCacheQueue.Free;
    FLockObject.Free;
    inherited Destroy;
end;

procedure TStreamingFileRecordReader<T>.ScanDirectoryAndMatchFiles;
var
    allFiles: TStringDynArray;
    fileNameOnly, filePath: string;
    expectedPrefix, baseExtensionPattern, zipExtensionPattern: string;
begin
    TMonitor.Enter(FLockObject);
    try
        FFilePaths.Clear;
        allFiles := TDirectory.GetFiles(FDirectoryPath);
        expectedPrefix := FDataSetName + '_';
        baseExtensionPattern := '.' + FExtension;
        zipExtensionPattern := baseExtensionPattern + '_zip';

        for filePath in allFiles do
        begin
            fileNameOnly := ExtractFileName(filePath);
            if StartsText(expectedPrefix, fileNameOnly) then
            begin
                if EndsText(zipExtensionPattern, fileNameOnly.ToLower) or EndsText(baseExtensionPattern, fileNameOnly.ToLower) then
                begin
                    FFilePaths.Add(filePath);
                end;
            end;
        end;
    finally
        TMonitor.Exit(FLockObject);
    end;
end;

function TStreamingFileRecordReader<T>.LoadPage(ActualFilePath: string): IDataPage;
var
    dataStream: TMemoryStream;
    fileLastModified: TDateTime;
    zipFile: TZipFile;
    isCompressed: Boolean;
    fileExtensionOnDisk: string;
    decompressionStream: TStream;
    localHeader: TZipHeader;
    fileNameInZip: string;
begin
    dataStream := TMemoryStream.Create;
    try
        fileLastModified := TFile.GetLastWriteTime(ActualFilePath);
        fileExtensionOnDisk := ExtractFileExt(ActualFilePath);
        isCompressed := EndsText('_zip', fileExtensionOnDisk.ToLower);

        if isCompressed then
        begin
            zipFile := TZipFile.Create;
            try
                zipFile.Open(ActualFilePath, zmRead);
                if zipFile.FileCount > 0 then
                begin
                    fileNameInZip := zipFile.FileNames[0];
                    zipFile.Read(fileNameInZip, decompressionStream, localHeader);
                    try
                        dataStream.CopyFrom(decompressionStream, 0);
                    finally
                        decompressionStream.Free;
                    end;
                end
                else
                    raise Exception.Create('Compressed file is empty: ' + ActualFilePath);
            finally
                zipFile.Free;
            end;
        end
        else
            dataStream.LoadFromFile(ActualFilePath);

        dataStream.Position := 0;
        result := TLocalFilePage.Create(ActualFilePath, dataStream, fileLastModified);
    except
        dataStream.Free;
        raise;
    end;
end;

procedure TStreamingFileRecordReader<T>.EnsurePageInCache(ActualFilePath: string; IsPreload: Boolean);
var
    page: IDataPage;
    newlyLoadedPage: IDataPage;
begin
    // Assumes FLockObject is held by the caller
    if not FCachedPages.TryGetValue(ActualFilePath, page) then
    begin
        if FCachedPages.Count >= FMaxCacheSize then
            EvictOldestPage;
        newlyLoadedPage := LoadPage(ActualFilePath);
        FCachedPages.Add(ActualFilePath, newlyLoadedPage);
        FPageCacheQueue.Enqueue(ActualFilePath);
    end
    else
    begin
        if not IsPreload then
            InvalidatePageIfModified(page, page = FCurrentPage);
    end;
end;

procedure TStreamingFileRecordReader<T>.EvictOldestPage;
var
    oldestFilePath: string;
begin
    // Assumes FLockObject is held by the caller
    if FPageCacheQueue.Count > 0 then
    begin
        oldestFilePath := FPageCacheQueue.Dequeue;
        if FCachedPages.ContainsKey(oldestFilePath) then
            FCachedPages.Remove(oldestFilePath);
    end;
end;

procedure TStreamingFileRecordReader<T>.InvalidatePageIfModified(PageToCheck: IDataPage; IsCurrentPage: Boolean);
var
    actualFilePath: string;
    reloadedPage: IDataPage;
begin
    // Assumes FLockObject is held by the caller
    if PageToCheck = nil then exit;

    actualFilePath := PageToCheck.GetSourceIdentifier;

    if TLocalFilePage.IsFileModified(actualFilePath, PageToCheck.GetLastModificationTime) then
    begin
        FCachedPages.Remove(actualFilePath);

        if IsCurrentPage then
        begin
            FCurrentPage := nil;
            FCurrentPageStream := nil;

            reloadedPage := LoadPage(actualFilePath);
            FCachedPages.Add(actualFilePath, reloadedPage);

            FCurrentPage := reloadedPage;
            FCurrentPageStream := FCurrentPage.GetStream;
            if (FCurrentPageStream <> nil) and (FCurrentPageStream.Size >= FRecordSize) then
                FRecordsInCurrentPage := FCurrentPageStream.Size div FRecordSize
            else
                FRecordsInCurrentPage := 0;

            if (FRecordsReadFromCurrentPage * FRecordSize < FCurrentPageStream.Size) and (FRecordsReadFromCurrentPage <= FRecordsInCurrentPage) then
                FCurrentPageStream.Position := FRecordsReadFromCurrentPage * FRecordSize
            else
            begin
                FCurrentPageStream.Position := 0;
                FRecordsReadFromCurrentPage := 0;
            end;
        end;
    end;
end;

procedure TStreamingFileRecordReader<T>.DoPreloadWork(const APathToPreload: string);
begin
    try
        TMonitor.Enter(FLockObject);
        try
            if not FCachedPages.ContainsKey(APathToPreload) then
                EnsurePageInCache(APathToPreload, True);
        finally
            TMonitor.Exit(FLockObject);
        end;
    except
        on E: Exception do
        begin
            // Optional: Log error E.Message
        end;
    end;
end;

procedure TStreamingFileRecordReader<T>.StartPreloadNextPage;
var
    nextPageIndexToLoad: Integer;
    nextFilePath: string;
    currentTask: ITask;
begin
    TMonitor.Enter(FLockObject);
    try
        currentTask := FPreloadPageTask;
        if (currentTask <> nil) and (currentTask.Status in [TTaskStatus.Created, TTaskStatus.WaitingToRun, TTaskStatus.Running]) then
            exit;

        nextPageIndexToLoad := FCurrentPageIndexInFileList + 1;
        if nextPageIndexToLoad < FFilePaths.Count then
        begin
            nextFilePath := FFilePaths[nextPageIndexToLoad];
            if FCachedPages.ContainsKey(nextFilePath) then
                exit;

            FPreloadPageTask := TTask.Run(
                procedure
                begin
                    DoPreloadWork(nextFilePath);
                end
            );
        end;
    finally
        TMonitor.Exit(FLockObject);
    end;
end;

function TStreamingFileRecordReader<T>.ActivatePage(PageIndexInFileList: Integer): Boolean;
var
    actualFilePath: string;
    pageFromCache: IDataPage;
begin
    result := False;
    if (PageIndexInFileList < 0) or (PageIndexInFileList >= FFilePaths.Count) then
    begin FCurrentPage := nil; FCurrentPageStream := nil; exit; end;

    actualFilePath := FFilePaths[PageIndexInFileList];

    TMonitor.Enter(FLockObject);
    try
        EnsurePageInCache(actualFilePath, False);
        if not FCachedPages.TryGetValue(actualFilePath, pageFromCache) then
        begin FCurrentPage := nil; FCurrentPageStream := nil; exit; end;

        FCurrentPage := pageFromCache;
        FCurrentPageStream := FCurrentPage.GetStream;
        FCurrentPageIndexInFileList := PageIndexInFileList;
        FRecordsReadFromCurrentPage := 0;
        if (FCurrentPageStream <> nil) and (FCurrentPageStream.Size >= FRecordSize) then
            FRecordsInCurrentPage := FCurrentPageStream.Size div FRecordSize
        else
            FRecordsInCurrentPage := 0;

        if FRecordsInCurrentPage > 0 then
        begin
            result := True;
            StartPreloadNextPage;
        end;
    finally
        TMonitor.Exit(FLockObject);
    end;
end;

function TStreamingFileRecordReader<T>.GetRecordCountForFile(ActualFilePath: string): Int64;
var
  page: IDataPage;
  stream: TStream;
begin
  result := 0;
  EnsurePageInCache(ActualFilePath, False); // Assumes FLockObject is held by caller
  if FCachedPages.TryGetValue(ActualFilePath, page) then
  begin
    stream := page.GetStream;
    if (stream <> nil) and (stream.Size >= FRecordSize) then
      result := stream.Size div FRecordSize;
  end;
end;

function TStreamingFileRecordReader<T>.GetCurrentRecord: T;
begin
    if FGlobalRecordIndex < 0 then
      Result := System.Default(T)
    else
      Result := FCurrentRecord;
end;

function TStreamingFileRecordReader<T>.GetCurrentRecordIndex: Int64;
begin
    result := FGlobalRecordIndex;
end;

function TStreamingFileRecordReader<T>.MoveToNextRecord: Boolean;
var
    bytesRead: NativeInt; // ReadData<T> returns NativeInt
    attemptedNextPageActivation: Boolean;
begin
    result := False;
    attemptedNextPageActivation := False;

    TMonitor.Enter(FLockObject);
    try
        while True do
        begin
            if FCurrentPageStream = nil then
            begin
                if attemptedNextPageActivation or (FCurrentPageIndexInFileList >= FFilePaths.Count - 1) then
                begin FGlobalRecordIndex := -1; exit; end;
                if not ActivatePage(FCurrentPageIndexInFileList + 1) then
                begin FGlobalRecordIndex := -1; exit; end;
                attemptedNextPageActivation := True;
            end;

            if FRecordsReadFromCurrentPage >= FRecordsInCurrentPage then
            begin
                if attemptedNextPageActivation or (FCurrentPageIndexInFileList >= FFilePaths.Count - 1) then
                begin FGlobalRecordIndex := -1; exit; end;
                FCurrentPageStream := nil;
                if not ActivatePage(FCurrentPageIndexInFileList + 1) then
                begin FGlobalRecordIndex := -1; exit; end;
                attemptedNextPageActivation := True;
                continue;
            end;

            FCurrentPageStream.Position := FRecordsReadFromCurrentPage * FRecordSize;
            if (FCurrentPageStream.Size - FCurrentPageStream.Position) < FRecordSize then
            begin
                FRecordsReadFromCurrentPage := FRecordsInCurrentPage;
                attemptedNextPageActivation := False;
                FCurrentPageStream := nil;
                continue;
            end;

            // Verwende die generische TStream.ReadData<T> Methode
            bytesRead := FCurrentPageStream.ReadData<T>(FCurrentRecord); // User line 518 context

            if bytesRead = FRecordSize then // FRecordSize is SizeOf(T)
            begin
                Inc(FRecordsReadFromCurrentPage);
                if FGlobalRecordIndex < 0 then
                    FGlobalRecordIndex := 0
                else
                    Inc(FGlobalRecordIndex); // << User line 518 (E2010), hopefully fixed by ReadData<T>
                result := True; exit;
            end
            else
            begin
                FRecordsReadFromCurrentPage := FRecordsInCurrentPage;
                attemptedNextPageActivation := False;
                FCurrentPageStream := nil;
                continue;
            end;
        end;
    finally
        TMonitor.Exit(FLockObject);
    end;
end;

procedure TStreamingFileRecordReader<T>.Reset;
var
    taskToCheck: ITask;
begin
    TMonitor.Enter(FLockObject);
    try
        taskToCheck := FPreloadPageTask;
        if (taskToCheck <> nil) and (taskToCheck.Status in [TTaskStatus.Created, TTaskStatus.WaitingToRun, TTaskStatus.Running]) then
        begin
            // Cancellation logic
        end;
        FPreloadPageTask := nil;

        FCurrentPageIndexInFileList := -1;
        FGlobalRecordIndex := -1;
        FRecordsReadFromCurrentPage := 0;
        FRecordsInCurrentPage := 0;
        FCurrentPage := nil;
        FCurrentPageStream := nil;
        FCurrentRecord := System.Default(T);
        if FFilePaths.Count > 0 then
            if not ActivatePage(0) then;
    finally
        TMonitor.Exit(FLockObject);
    end;
end;

function TStreamingFileRecordReader<T>.Seek(RecordIndex: Int64): Int64;
var
    pageIdx: Integer;
    cumulativeRecordOffset: Int64;
    recordsInCurrentFile: Int64;
    recordOffsetInPage: Int64;
    effectiveTargetIdx: Int64;
    i: Integer;
    sumRecordsBeforeThisPage: Int64;
    bytesRead: NativeInt; // For ReadData<T>
begin
    TMonitor.Enter(FLockObject);
    try
        if FFilePaths.Count = 0 then
        begin
            FCurrentPageIndexInFileList := -1; FGlobalRecordIndex := -1; FCurrentPage := nil; FCurrentPageStream := nil;
            FCurrentRecord := System.Default(T);
            result := -1; exit;
        end;

        effectiveTargetIdx := RecordIndex;
        if effectiveTargetIdx < 0 then effectiveTargetIdx := 0;

        cumulativeRecordOffset := 0;
        for pageIdx := 0 to FFilePaths.Count - 1 do
        begin
            recordsInCurrentFile := GetRecordCountForFile(FFilePaths[pageIdx]);
            if effectiveTargetIdx < cumulativeRecordOffset + recordsInCurrentFile then
            begin
                if not ActivatePage(pageIdx) then
                begin
                    FCurrentPageIndexInFileList := -1; FGlobalRecordIndex := -1; FCurrentPage := nil; FCurrentPageStream := nil;
                    FCurrentRecord := System.Default(T);
                    result := -1; exit;
                end;
                recordOffsetInPage := effectiveTargetIdx - cumulativeRecordOffset;
                FCurrentPageStream.Position := recordOffsetInPage * FRecordSize;

                // Verwende die generische TStream.ReadData<T> Methode
                bytesRead := FCurrentPageStream.ReadData<T>(FCurrentRecord);

                if bytesRead = FRecordSize then
                begin
                    FGlobalRecordIndex := effectiveTargetIdx; FRecordsReadFromCurrentPage := recordOffsetInPage + 1;
                    result := FGlobalRecordIndex; exit;
                end
                else
                begin // User line ~604 context (E2008)
                    FCurrentPageIndexInFileList := -1; FGlobalRecordIndex := -1; FCurrentPage := nil; FCurrentPageStream := nil;
                    FCurrentRecord := System.Default(T);
                    result := -1; exit;
                end;
            end;
            inc(cumulativeRecordOffset, recordsInCurrentFile);
        end;

        if cumulativeRecordOffset = 0 then // User line ~616 context (E2029) for assignment after this block
        begin
            FCurrentPageIndexInFileList := -1; FGlobalRecordIndex := -1; FCurrentPage := nil; FCurrentPageStream := nil;
            FCurrentRecord := System.Default(T);
            result := -1; exit;
        end;

        for pageIdx := FFilePaths.Count - 1 downto 0 do
        begin
            recordsInCurrentFile := GetRecordCountForFile(FFilePaths[pageIdx]);
            if recordsInCurrentFile > 0 then
            begin
                if not ActivatePage(pageIdx) then
                begin
                    FCurrentPageIndexInFileList := -1; FGlobalRecordIndex := -1; FCurrentPage := nil; FCurrentPageStream := nil;
                    FCurrentRecord := System.Default(T);
                    result := -1; exit;
                end;
                recordOffsetInPage := recordsInCurrentFile - 1;
                FCurrentPageStream.Position := recordOffsetInPage * FRecordSize;

                // Verwende die generische TStream.ReadData<T> Methode
                bytesRead := FCurrentPageStream.ReadData<T>(FCurrentRecord);

                if bytesRead = FRecordSize then
                begin
                    sumRecordsBeforeThisPage := 0;
                    for i := 0 to pageIdx - 1 do
                        sumRecordsBeforeThisPage := sumRecordsBeforeThisPage + GetRecordCountForFile(FFilePaths[i]);
                    FGlobalRecordIndex := sumRecordsBeforeThisPage + recordOffsetInPage; // User line 616 context (E2029/E2014)
                    FRecordsReadFromCurrentPage := recordOffsetInPage + 1;
                    result := FGlobalRecordIndex; exit;
                end
                else
                begin // User line ~639 context (E2008)
                    FCurrentPageIndexInFileList := -1; FGlobalRecordIndex := -1; FCurrentPage := nil; FCurrentPageStream := nil;
                    FCurrentRecord := System.Default(T);
                    result := -1; exit;
                end;
            end;
        end;
        FCurrentPageIndexInFileList := -1; FGlobalRecordIndex := -1; FCurrentPage := nil; FCurrentPageStream := nil;
        FCurrentRecord := System.Default(T);
        result := -1; exit;
    finally
        TMonitor.Exit(FLockObject);
    end;
end;

end.