integrated TFuture
This commit is contained in:
@@ -0,0 +1,671 @@
|
||||
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.
|
||||
Reference in New Issue
Block a user