From f8c3ffceb8c8a1238bc17b368b7b81a8cb8ee26e Mon Sep 17 00:00:00 2001 From: Michael Schimmel Date: Thu, 5 Jun 2025 14:36:05 +0200 Subject: [PATCH] TDataSeries --- Src/Myc.Core.Futures.pas | 32 +- Src/Myc.Core.Notifier.pas | 2 +- Src/Myc.Core.Tasks.pas | 22 +- Src/Myc.Test.Core.Atomic.pas | 6 +- Src/Myc.Test.Core.Lazy.pas | 124 +++---- Src/Myc.Test.Lazy.pas | 108 +++--- Src/Myc.Trade.DataSeries.pas | 542 ++++++++++++++++++++++++++++++ Test/TestCoreFutures.pas | 166 ++++----- Test/TestFutures.pas | 316 ++++++++--------- Test/TestNotifier.pas | 34 +- Test/TestNotifier_ChaosStress.pas | 59 ++-- Test/TestNotifier_Threading.pas | 40 +-- Test/TestSignals_Latch.pas | 25 +- Test/TestTasks.pas | 92 ++--- 14 files changed, 1051 insertions(+), 517 deletions(-) create mode 100644 Src/Myc.Trade.DataSeries.pas diff --git a/Src/Myc.Core.Futures.pas b/Src/Myc.Core.Futures.pas index 59a88d3..2acb419 100644 --- a/Src/Myc.Core.Futures.pas +++ b/Src/Myc.Core.Futures.pas @@ -59,22 +59,22 @@ begin // Subscribe the job execution to AGate. // The job will run when AGate notifies the subscriber returned by Run. FInit := - ATaskManager.CreateTask( - AGate, - procedure - begin - try - try - Self.FResult := AProc(); - except - Self.FResult := Default(T); // Set result to Default(T) on error - raise; // Re-raise for TaskFactory to handle - end; - finally - Self.FDone.Notify; // Signal that this future is done (successfully or with error) - end; - end - ); + ATaskManager.CreateTask( + AGate, + procedure + begin + try + try + Self.FResult := AProc(); + except + Self.FResult := Default(T); // Set result to Default(T) on error + raise; // Re-raise for TaskFactory to handle + end; + finally + Self.FDone.Notify; // Signal that this future is done (successfully or with error) + end; + end + ); end; destructor TMycGateFuncFuture.Destroy; diff --git a/Src/Myc.Core.Notifier.pas b/Src/Myc.Core.Notifier.pas index f82b5a9..9b6408a 100644 --- a/Src/Myc.Core.Notifier.pas +++ b/Src/Myc.Core.Notifier.pas @@ -43,7 +43,7 @@ type procedure Release; inline; // Releases the previously acquired exclusive lock. function IsLocked: Boolean; inline; // Checks if the list is currently locked by any thread. procedure Notify( - Func: TPredicate + Func: TPredicate ); // Iterates through registered receivers and invokes the predicate; removes receiver if predicate returns false. end; diff --git a/Src/Myc.Core.Tasks.pas b/Src/Myc.Core.Tasks.pas index 3ba6d38..eb5ecea 100644 --- a/Src/Myc.Core.Tasks.pas +++ b/Src/Myc.Core.Tasks.pas @@ -211,17 +211,17 @@ begin capturedProc := Proc; // Capture Proc for the anonymous method CreateAnonymousThread( - 'Thread', - procedure - begin - try - capturedProc(); // Execute the provided procedure - finally - capturedProc := nil; // Clear the captured proc - res.Notify; // Signal completion via the latch's Notify method - end; - end) - .Start; + 'Thread', + procedure + begin + try + capturedProc(); // Execute the provided procedure + finally + capturedProc := nil; // Clear the captured proc + res.Notify; // Signal completion via the latch's Notify method + end; + end) + .Start; // Return the state interface of the latch exit(res.State); diff --git a/Src/Myc.Test.Core.Atomic.pas b/Src/Myc.Test.Core.Atomic.pas index 4863736..a1d77b8 100644 --- a/Src/Myc.Test.Core.Atomic.pas +++ b/Src/Myc.Test.Core.Atomic.pas @@ -279,9 +279,9 @@ begin entryData := AllocEntryData(123); // This should not raise an assertion error from TSList.Push. Assert.WillNotRaise( - procedure begin FList.Push(@entryData.Entry); end, - nil, - 'Pushing an aligned entry should not raise an exception/assertion.' + procedure begin FList.Push(@entryData.Entry); end, + nil, + 'Pushing an aligned entry should not raise an exception/assertion.' ); Assert.AreEqual(1, FList.QueryDepth, 'QueryDepth should be 1 after pushing an aligned entry.'); end; diff --git a/Src/Myc.Test.Core.Lazy.pas b/Src/Myc.Test.Core.Lazy.pas index f41cf38..256065b 100644 --- a/Src/Myc.Test.Core.Lazy.pas +++ b/Src/Myc.Test.Core.Lazy.pas @@ -187,14 +187,14 @@ begin procExecuted := False; expectedValue := 50; funcLazy := - TMycFuncLazy.Create( - sourceDirty.State, - function: Integer - begin - procExecuted := True; - Result := expectedValue; - end - ); + TMycFuncLazy.Create( + sourceDirty.State, + function: Integer + begin + procExecuted := True; + Result := expectedValue; + end + ); Assert.IsNotNull(funcLazy, 'TMycFuncLazy instance should not be nil'); Assert.IsTrue(funcLazy.GetChanged.IsSet, 'Changed.IsSet should be true before the first Pop by design'); value := 0; @@ -223,9 +223,9 @@ begin result := funcLazy.Pop(value); Assert.IsTrue(result, 'First Pop should return true by design, even if FProc is nil'); Assert.AreEqual( - preCallValue, - value, - 'Value should be unchanged as FProc was nil and current Pop implementation does not assign Default(T)' + preCallValue, + value, + 'Value should be unchanged as FProc was nil and current Pop implementation does not assign Default(T)' ); Assert.IsFalse(funcLazy.GetChanged.IsSet, 'Changed state should be false after Pop'); end; @@ -260,14 +260,14 @@ begin initialProcValue := 44; procExecutedCount := 0; funcLazy := - TMycFuncLazy.Create( - sourceDirty.State, - function: Integer - begin - Inc(procExecutedCount); - Result := initialProcValue; - end - ); + TMycFuncLazy.Create( + sourceDirty.State, + function: Integer + begin + Inc(procExecutedCount); + Result := initialProcValue; + end + ); Helper_ConsumeInitialPop(funcLazy, initialProcValue); Assert.AreEqual(1, procExecutedCount, 'Proc should have executed once for initial pop'); result := funcLazy.Pop(value); @@ -305,17 +305,17 @@ begin sourceDirty.Reset; procCallCount := 0; funcLazy := - TMycFuncLazy.Create( - sourceDirty.State, - function: Integer - begin - Inc(procCallCount); - if procCallCount = 1 then - Result := 30 - else - Result := 300 + procCallCount; - end - ); + TMycFuncLazy.Create( + sourceDirty.State, + function: Integer + begin + Inc(procCallCount); + if procCallCount = 1 then + Result := 30 + else + Result := 300 + procCallCount; + end + ); currentExpectedValue := 30; Helper_ConsumeInitialPop(funcLazy, currentExpectedValue); Assert.AreEqual(1, procCallCount, 'Proc executed for initial Pop'); @@ -342,14 +342,14 @@ begin sourceDirty.Reset; procCallCount := 0; funcLazy := - TMycFuncLazy.Create( - sourceDirty.State, - function: Integer - begin - Inc(procCallCount); - Result := 40; - end - ); + TMycFuncLazy.Create( + sourceDirty.State, + function: Integer + begin + Inc(procCallCount); + Result := 40; + end + ); resultPop1 := funcLazy.Pop(value); Assert.IsTrue(resultPop1, 'First Pop should return true by design'); Assert.AreEqual(40, value, 'Value from first Pop'); @@ -378,19 +378,19 @@ begin firstSourceChangeValue := 52; funcLazy := - TMycFuncLazy.Create( - sourceDirty.State, - function: Integer - begin - Inc(procCallCount); - if procCallCount = 1 then - Result := initialValue // For initial pop - else if procCallCount = 2 then - Result := firstSourceChangeValue // For pop after first effective source change - else - Result := 999; // Should not be reached in this specific test logic - end - ); + TMycFuncLazy.Create( + sourceDirty.State, + function: Integer + begin + Inc(procCallCount); + if procCallCount = 1 then + Result := initialValue // For initial pop + else if procCallCount = 2 then + Result := firstSourceChangeValue // For pop after first effective source change + else + Result := 999; // Should not be reached in this specific test logic + end + ); // 1. Initial Pop (consumes "by design" changed state) Assert.IsTrue(funcLazy.GetChanged.IsSet, 'Changed state should be true before first Pop (by design)'); @@ -436,9 +436,9 @@ begin Assert.IsTrue(funcLazyObj.Pop(tempVal), 'Initial Pop should succeed'); funcLazyObj.Destroy; Assert.WillNotRaise( - procedure begin sourceDirty.Notify; end, - nil, - 'Destroy test assumes TMycSubscription.Unsubscribe works. Verified by no crash on source notify post-destroy.' + procedure begin sourceDirty.Notify; end, + nil, + 'Destroy test assumes TMycSubscription.Unsubscribe works. Verified by no crash on source notify post-destroy.' ); sourceDirty := nil; end; @@ -457,14 +457,14 @@ begin expectedValue := 70; procExecuted := False; funcLazy := - TMycFuncLazy.Create( - sourceDirty.State, - function: Integer - begin - procExecuted := True; - Result := expectedValue; - end - ); + TMycFuncLazy.Create( + sourceDirty.State, + function: Integer + begin + procExecuted := True; + Result := expectedValue; + end + ); Assert.IsNotNull(funcLazy, 'TMycFuncLazy instance should not be nil'); Assert.IsTrue(funcLazy.GetChanged.IsSet, 'funcLazy.GetChanged.IsSet should be true after creation (by design)'); value := 0; diff --git a/Src/Myc.Test.Lazy.pas b/Src/Myc.Test.Lazy.pas index 40ed255..bc41443 100644 --- a/Src/Myc.Test.Lazy.pas +++ b/Src/Myc.Test.Lazy.pas @@ -146,14 +146,14 @@ begin procExecuted := False; // FChangingSignal is reset in Setup lazyIntf := - TLazy.Construct( - FChangingSignal.State, - function: Integer - begin - procExecuted := True; - Result := 10; - end - ); + TLazy.Construct( + FChangingSignal.State, + function: Integer + begin + procExecuted := True; + Result := 10; + end + ); Assert.IsNotNull(lazyIntf, 'TLazy.Construct should return a valid interface'); lazyRec := lazyIntf; // Implicit conversion @@ -171,14 +171,14 @@ begin procExecuted := False; expectedValue := 20; lazyIntf := - TLazy.Construct( - FChangingSignal.State, - function: Integer - begin - procExecuted := True; - Result := expectedValue; - end - ); + TLazy.Construct( + FChangingSignal.State, + function: Integer + begin + procExecuted := True; + Result := expectedValue; + end + ); lazyRec := lazyIntf; ConsumeInitialPop(lazyRec, expectedValue, 'TestConstruct_FirstPop'); @@ -236,14 +236,14 @@ var begin procCallCount := 0; lazyIntf := - TLazy.Construct( - FChangingSignal.State, - function: Integer - begin - Inc(procCallCount); - Result := 100 + procCallCount; // Value changes per call - end - ); + TLazy.Construct( + FChangingSignal.State, + function: Integer + begin + Inc(procCallCount); + Result := 100 + procCallCount; // Value changes per call + end + ); lazyRec := lazyIntf; ConsumeInitialPop(lazyRec, 101, 'TestConstruct_StateInteraction_PopResetsChangedAfterSignal (Initial)'); // procCallCount = 1 @@ -267,14 +267,14 @@ var begin procCallCount := 0; lazyIntf := - TLazy.Construct( - FChangingSignal.State, - function: Integer - begin - Inc(procCallCount); - Result := 200 + procCallCount; - end - ); + TLazy.Construct( + FChangingSignal.State, + function: Integer + begin + Inc(procCallCount); + Result := 200 + procCallCount; + end + ); lazyRec := lazyIntf; // 1. Initial Pop @@ -322,12 +322,12 @@ begin // This should trigger its destructor, which should unsubscribe from localChangingSignal. Assert.WillNotRaise( - procedure - begin - localChangingSignal.Notify; // Notify the source AFTER the lazy object is supposed to be gone. - end, - nil, // Default: any exception is a failure - 'Notifying source after lazy object is freed should not crash, indicating unsubscription.' + procedure + begin + localChangingSignal.Notify; // Notify the source AFTER the lazy object is supposed to be gone. + end, + nil, // Default: any exception is a failure + 'Notifying source after lazy object is freed should not crash, indicating unsubscription.' ); localChangingSignal := nil; // Clean up the local signal itself. @@ -368,14 +368,14 @@ var begin procCallCount := 0; originalLazyIntf := - TLazy.Construct( - FChangingSignal.State, - function: Integer - begin - Inc(procCallCount); - Result := 70 + procCallCount; - end - ); + TLazy.Construct( + FChangingSignal.State, + function: Integer + begin + Inc(procCallCount); + Result := 70 + procCallCount; + end + ); wrappedLazyRec := TLazy.Create(originalLazyIntf); // First Pop via wrapper (initial pop) @@ -450,14 +450,14 @@ var begin procExecuted := False; lazyIntf := - TLazy.Construct( - FChangingSignal.State, - function: Integer - begin - procExecuted := True; - Result := 100; - end - ); + TLazy.Construct( + FChangingSignal.State, + function: Integer + begin + procExecuted := True; + Result := 100; + end + ); lazyRec := lazyIntf; ConsumeInitialPop(lazyRec, 100, 'TestPop_AfterInitialAndNoSignal (Initial)'); diff --git a/Src/Myc.Trade.DataSeries.pas b/Src/Myc.Trade.DataSeries.pas new file mode 100644 index 0000000..a882f14 --- /dev/null +++ b/Src/Myc.Trade.DataSeries.pas @@ -0,0 +1,542 @@ +unit Myc.Trade.DataSeries; + +(* ------------------------------------------------------------------------------ + This unit provides functions to load tick data (TTickData) from + binary files. The data originates from a C# application + and has the structure OADateTime (Double), Ask (Single), and Bid (Single). + + The files follow the naming convention: + For .tab files: Symbol_Year_MM.tab OR Symbol_Year_MM.tab_zip + For .tab-live files: Symbol_Year_MM.tab-live (NEVER compressed) + (MM represents the two-digit month, e.g., 01 for January, 12 for December) + + If both compressed (.tab_zip) and uncompressed (.tab) versions of a .tab + file exist, the compressed version is preferred. Files ending with _zip + are expected to be ZIP archives containing the actual data file + (e.g., Symbol_Year_MM.tab). + + Loading occurs in two phases: + 1. All .tab files (regular or _zip) are processed chronologically by + month and year. Unique ticks are loaded, and the timestamp of the + latest tick across all .tab files is determined (overallLastTabTickTime). + Assumes timestamps are unique across all .tab files. + 2. All .tab-live files (which are never compressed), for months/years + corresponding to found .tab files, are processed. Ticks from a + .tab-live file are only integrated if their timestamp is strictly later + than overallLastTabTickTime. Assumes such ticks are globally unique. + + If a .tab-live file results in no new data being added under these rules, + it is deleted after processing. + + The loading function automatically detects subsequent monthly/yearly files. + + Main Function: + LoadTickDataSeries - Loads a series of tick data files. + ------------------------------------------------------------------------------ *) + +interface + +uses + System.SysUtils, + System.Classes, + System.Generics.Collections, + System.IOUtils, + Myc.Futures, + Myc.Signals; + +type + TDataSeries = class + type + TDataPoint = packed record + OADateTime: Double; + Data: T; + end; + protected + class function ReadCompressedData(const InputStream: TStream): TArray; static; + class function ReadUncompressedData(const InputStream: TStream): TArray; static; + public + class function LoadDataFile(var LoadGate: TLatch; const FileName: string): TFuture>; static; + class function LoadDataSeries(const FileName: string): TFuture>; static; + end; + + TAskBidItem = packed record + Ask: Single; + Bid: Single; + end; + + TAskBid = class(TDataSeries) + type + TTick = TDataSeries.TDataPoint; + end; + +function TryParseFileName(const FileName: string; out PathValue, SymbolValue: string; out YearValue, MonthValue: Integer): Boolean; + +// Finds the full filename of the oldest (=first) file for each symbol in the path. +function FindOldestFilesPerSymbol(const DirectoryPath: string): TArray; + +implementation + +uses + System.Zip, + Myc.TaskManager; + +// TryParseFileName parses filenames like "Symbol_Year_MM.Extension" or "Symbol_Year_MM.Extension_zip" +// to extract Path, Symbol, Year, and Month. +function TryParseFileName(const FileName: string; out PathValue, SymbolValue: string; out YearValue, MonthValue: Integer): Boolean; +var + fileNameNoPath: string; + nameForParsing: string; + parts: TArray; + baseName: string; +begin + Result := False; // Default to failure + PathValue := ''; // Initialize all out parameters + SymbolValue := ''; + YearValue := 0; + MonthValue := 0; + + PathValue := TPath.GetDirectoryName(FileName); + fileNameNoPath := TPath.GetFileName(FileName); + + nameForParsing := fileNameNoPath; + // Handle cases like "Symbol_Year_MM.Extension_zip" by removing the trailing "_zip" before extension removal + if nameForParsing.EndsWith('_zip', True) then + begin + // Get the part before "_zip" + baseName := nameForParsing.Substring(0, nameForParsing.Length - 4); + // Get the extension of the original name (e.g. ".csv_zip") + // and check if baseName still contains an extension that needs to be removed by GetFileNameWithoutExtension + if TPath.GetExtension(baseName) <> '' then + begin + nameForParsing := baseName; + end + else + begin + // If baseName (e.g. "EURUSD_2023_01") has no extension, it is the correct parsing base + // However, TPath.GetFileNameWithoutExtension would also work correctly here, + // but this handles a potential case where a file might be named "Symbol_Year_MM_zip" (no dot before zip) + end; + end; + + // Remove the actual extension (e.g., ".csv", ".dat") from "Symbol_Year_MM.Extension" or the modified "Symbol_Year_MM.Extension" part of "_zip" + nameForParsing := TPath.GetFileNameWithoutExtension(nameForParsing); // e.g., "EURUSD_2023_01" + + parts := nameForParsing.Split(['_']); + + if Length(parts) < 3 then // Need at least Symbol_Year_Month + exit; + + // Month is the last part. + if not TryStrToInt(parts[High(parts)], MonthValue) then + begin + exit; + end; + + if (MonthValue < 1) or (MonthValue > 12) then // Validate month range + begin + MonthValue := 0; // Set to invalid if out of range + exit; + end; + + // Year is the second to last part. + if not TryStrToInt(parts[High(parts) - 1], YearValue) then + begin + MonthValue := 0; // Also invalidate month if year parsing fails + exit; + end; + // Ensure YearValue is plausible, e.g. not 0 or negative, adjust as needed for your data's year range + if YearValue <= 0 then + begin + YearValue := 0; + MonthValue := 0; + exit; + end; + + // Symbol consists of all parts before Year and Month. + SymbolValue := string.Join('_', Copy(parts, 0, Length(parts) - 2)); + + if SymbolValue = '' then // Ensure symbol is not empty + begin + // Reset all out parameters on failure before exit + PathValue := ''; + YearValue := 0; + MonthValue := 0; + exit; + end; + + Result := True; +end; + +// Finds the oldest file for each symbol in the given directory. +// The file extension is ignored for the purpose of identifying the symbol, year, and month. +function FindOldestFilesPerSymbol(const DirectoryPath: string): TArray; +type + TTrackedFileInfo = record + FullFileName: string; + Year: Integer; + Month: Integer; + end; +var + oldestFilesPerSymbol: TDictionary; + fileNames: TArray; + currentFile: string; + parsedPath: string; + parsedSymbol: string; + parsedYear: Integer; + parsedMonth: Integer; + trackedInfo: TTrackedFileInfo; +begin + oldestFilesPerSymbol := TDictionary.Create; // Dictionary to track the oldest file per symbol + try + if not TDirectory.Exists(DirectoryPath) then + begin + // Or raise an exception, depending on desired error handling + exit; + end; + + fileNames := TDirectory.GetFiles(DirectoryPath); // Get all files in the directory + + for currentFile in fileNames do + begin + // Attempt to parse the file name + if TryParseFileName(currentFile, parsedPath, parsedSymbol, parsedYear, parsedMonth) then + begin + // Check if this symbol is already tracked or if this file is older + if oldestFilesPerSymbol.TryGetValue(parsedSymbol, trackedInfo) then + begin + // If current file's year is less, or year is same and month is less, it's older + if (parsedYear < trackedInfo.Year) or ((parsedYear = trackedInfo.Year) and (parsedMonth < trackedInfo.Month)) then + begin + trackedInfo.FullFileName := currentFile; // Store full path to the file + trackedInfo.Year := parsedYear; + trackedInfo.Month := parsedMonth; + oldestFilesPerSymbol.AddOrSetValue(parsedSymbol, trackedInfo); // Update the entry for this symbol + end; + end + else + begin + // First time we see this symbol, so it's the oldest so far + trackedInfo.FullFileName := currentFile; // Store full path to the file + trackedInfo.Year := parsedYear; + trackedInfo.Month := parsedMonth; + oldestFilesPerSymbol.Add(parsedSymbol, trackedInfo); // Add new entry for this symbol + end; + end; + end; + + // Populate the result array with the full names of the oldest files + SetLength(Result, oldestFilesPerSymbol.Count); + var n := 0; + for trackedInfo in oldestFilesPerSymbol.Values do + begin + Result[n] := trackedInfo.FullFileName; + inc(n); + end; + finally + oldestFilesPerSymbol.Free; // Free the dictionary + end; +end; + +// Asynchronously reads tick data from a single specified file. +// Returns a TFuture that will provide an array of TAskBidSeries. +class function TDataSeries.LoadDataFile(var LoadGate: TLatch; const FileName: string): TFuture>; +var + parsedPath, parsedSymbol: string; + parsedYear, parsedMonth: Integer; +begin + Result := TFuture>.Null; + + // Attempt to parse year and month from the filename + if not TryParseFileName(FileName, parsedPath, parsedSymbol, parsedYear, parsedMonth) then + exit; + + if FileName.EndsWith('_zip', True) then + begin + var zipFileBytes := TFile.ReadAllBytes(FileName); // Read all bytes of the .zip file + if Length(zipFileBytes) = 0 then + exit; + + var capFileName := FileName; + Result := + TFuture + .Construct( + TLatch.Enqueue(LoadGate), + function: TBytes + begin + Result := TFile.ReadAllBytes(capFileName); // Read all bytes of the .zip file + end) + .Chain>( + function(bytes: TBytes): TArray + begin + var zipMemoryStream := TBytesStream.Create(zipFileBytes); // Create a memory stream to hold the zip file content + try + zipMemoryStream.Position := 0; + Result := ReadCompressedData(zipMemoryStream); + finally + zipMemoryStream.Free; + end; + end); + end + else if TFile.Exists(FileName) then + begin + var capFileName := FileName; + Result := + TFuture>.Construct( + function: TArray + begin + if TFile.Exists(capFileName) then + begin + var stream := TFileStream.Create(capFileName, fmOpenRead or fmShareDenyWrite); + try + try + Result := ReadUncompressedData(stream); + except + on E: EReadError do + begin + raise EReadError.CreateFmt('File %s: %s', [capFileName, E.Message]); + end; + end; + finally + stream.Free; + end; + end; + end + ); + end; +end; + +class function TDataSeries.LoadDataSeries(const FileName: string): TFuture>; +// Local type declaration for storing file information for phase 1 processing +type + TPhase1FileEntry = record + Path: string; + Year: Integer; + Month: Integer; + end; +var + initialYear, initialMonth: Integer; + currentTabYear, currentTabMonth: Integer; + pathName, symbolName: string; + tabFileEntryRecord: TPhase1FileEntry; // Variable to construct record instances + loadedState: TState; + loadedFiles: TArray>>; + liveData: TFuture>; + +begin // Start of LoadDataSeries + if not TryParseFileName(FileName, pathName, symbolName, initialYear, initialMonth) then + begin + raise EArgumentException.CreateFmt( + 'Invalid initial file name format: %s. Expected Symbol_Year_MM.Extension or Symbol_Year_MM.Extension_zip', + [FileName]); + end; + + currentTabYear := initialYear; + currentTabMonth := initialMonth; + var tabBaseNameFormat: string; + var compressedTabPath, uncompressedTabPath: string; + var actualTabFileToLoad: string; + var maxTabYearFound := -1; + var maxTabMonthFound := -1; + + var filesToLoadPhase1 := TList.Create; // Create list with the named record type + try + while True do + begin + tabBaseNameFormat := Format('%s_%.4d_%.2d.tab', [symbolName, currentTabYear, currentTabMonth]); + compressedTabPath := TPath.Combine(pathName, tabBaseNameFormat + '_zip'); + uncompressedTabPath := TPath.Combine(pathName, tabBaseNameFormat); + + actualTabFileToLoad := ''; + if TFile.Exists(compressedTabPath) then + begin + actualTabFileToLoad := compressedTabPath; + end + else if TFile.Exists(uncompressedTabPath) then + begin + actualTabFileToLoad := uncompressedTabPath; + end; + + if actualTabFileToLoad = '' then + break; // No more sequential .tab files + + // Construct the record and add it to the list + tabFileEntryRecord.Path := actualTabFileToLoad; + tabFileEntryRecord.Year := currentTabYear; + tabFileEntryRecord.Month := currentTabMonth; + filesToLoadPhase1.Add(tabFileEntryRecord); + + // Advance to the next month + Inc(currentTabMonth); + if currentTabMonth > 12 then + begin + currentTabMonth := 1; + Inc(currentTabYear); + end; + end; + + // Now load and process identified .tab files + var loadedFileList := TList>>.Create; + var loadStates := TList.Create; + try + var LoadGate: TLatch; + + for var tabFileEntry: TPhase1FileEntry in filesToLoadPhase1 do // Iterate with the correct type + begin + var data := LoadDataFile(LoadGate, tabFileEntry.Path); + + loadedFileList.Add(TFuture>.Create(data)); + loadStates.Add(data.Done); + + if tabFileEntry.Year > maxTabYearFound then + begin + maxTabYearFound := tabFileEntry.Year; + maxTabMonthFound := tabFileEntry.Month; + end + else if (tabFileEntry.Year = maxTabYearFound) and (tabFileEntry.Month > maxTabMonthFound) then + begin + maxTabMonthFound := tabFileEntry.Month; + end; + end; + + var liveFileBaseName := Format('%s_%d_%02d.tab-live', [symbolName, maxTabYearFound, maxTabMonthFound]); + liveData := LoadDataFile(LoadGate, TPath.Combine(pathName, liveFileBaseName)); + loadStates.Add(liveData.Done); + loadedFileList.Add(liveData); + + loadedState := TState.All(loadStates.ToArray); + loadedFiles := loadedFileList.ToArray; + finally + loadStates.Free; + loadedFileList.Free; + end; + finally + filesToLoadPhase1.Free; + end; + + Result := + TFuture>.Construct( + loadedState, + function: TArray + begin + var tickList := TList.Create; + try + var totalTicks := Length(liveData.Result); + var overallLastTabTickTime := 0.0; + + for var P in loadedFiles do + Inc(totalTicks, Length(P.Result)); + + tickList.Capacity := totalTicks; + + for var P in loadedFiles do + begin + if Length(P.Result) > 0 then + begin + for var iterTickData in P.Result do + if iterTickData.OADateTime > overallLastTabTickTime then + begin + tickList.Add(iterTickData); + overallLastTabTickTime := iterTickData.OADateTime; + end; + end; + end; + + Result := tickList.ToArray; + finally + tickList.Free; + end; + end + ); +end; + +class function TDataSeries.ReadCompressedData(const InputStream: TStream): TArray; +var + decompressionStream: TStream; // Holds uncompressed data of a single zip entry + localHeader: TZipHeader; + entryIndex: Integer; + i: Integer; + zipFileInstance: TZipFile; +begin + // Initialize Result record + SetLength(Result, 0); + + decompressionStream := nil; + zipFileInstance := nil; + + try + InputStream.Position := 0; // Reset stream position to the beginning for TZipFile + + // Open TZipFile from this memory stream + zipFileInstance := TZipFile.Create; + zipFileInstance.Open(InputStream, TZipMode.zmRead); // Open the zip from the memory stream + + if zipFileInstance.FileCount = 0 then + begin + // If TZipFile opens an empty or invalid stream successfully but finds no files + exit; // FileSuccessfullyLoaded remains False + end; + + entryIndex := -1; + for i := 0 to zipFileInstance.FileCount - 1 do + begin + if zipFileInstance.FileNames[i].EndsWith('.tab', True) then + begin + entryIndex := i; + break; + end; + end; + + if entryIndex = -1 then // If specific name not found, default to the first entry + begin + if zipFileInstance.FileCount > 0 then + entryIndex := 0 + else + exit; // Should not happen if FileCount > 0 check passed + end; + + // Decompress the specific entry from the in-memory zip file. + // TZipFile.Read creates and populates decompressionStream with uncompressed data. + zipFileInstance.Read(entryIndex, decompressionStream, localHeader); + + if not Assigned(decompressionStream) then + exit; // Exit if no valid stream could be prepared + + Result := ReadUncompressedData(decompressionStream); + finally + // Ensure all potentially created objects are freed + if Assigned(decompressionStream) then // Holds uncompressed data from a zip entry + decompressionStream.Free; + if Assigned(zipFileInstance) then + zipFileInstance.Free; + end; +end; + +class function TDataSeries.ReadUncompressedData(const InputStream: TStream): TArray; +var + fileSize: Int64; + recordCount: Integer; + bytesRead: Integer; +begin + // Initialize Result record + SetLength(Result, 0); + + InputStream.Position := 0; + + fileSize := InputStream.Size; + if fileSize = 0 then + exit; + + if (fileSize mod SizeOf(TDataPoint)) <> 0 then + raise EReadError.CreateFmt('File corrupted. Size %d is not a multiple of record size %d.', [fileSize, SizeOf(TDataPoint)]); + + recordCount := fileSize div SizeOf(TDataPoint); + if recordCount > 0 then + begin + SetLength(Result, recordCount); + bytesRead := InputStream.Read(Result[0], fileSize); + if bytesRead <> fileSize then + raise EReadError.CreateFmt('Read error. Expected to read %d bytes, but actually read %d bytes.', [fileSize, bytesRead]); + end; +end; + +end. diff --git a/Test/TestCoreFutures.pas b/Test/TestCoreFutures.pas index 052b391..2e4c49b 100644 --- a/Test/TestCoreFutures.pas +++ b/Test/TestCoreFutures.pas @@ -99,15 +99,15 @@ begin LInitStateAsState := TState.Null; LFuture := - TMycGateFuncFuture.Create( - FTaskFactory, - LInitStateAsState, - function: Integer - begin - Inc(Self.FProcExecutionCount); - Result := CExpectedResult; - end - ); + TMycGateFuncFuture.Create( + FTaskFactory, + LInitStateAsState, + function: Integer + begin + Inc(Self.FProcExecutionCount); + Result := CExpectedResult; + end + ); FTaskFactory.WaitFor(LFuture.Done); // Accesses IMycFuture.GetDone [cite: 110, 234, 352] @@ -129,15 +129,15 @@ begin LInitLatch := TLatch.Construct(1); // Create an init state that is not yet set [cite: 77, 212, 330] LFuture := - TMycGateFuncFuture.Create( - FTaskFactory, - LInitLatch.State, - function: string - begin - Inc(Self.FProcExecutionCount); - Result := CExpectedResult; - end - ); + TMycGateFuncFuture.Create( + FTaskFactory, + LInitLatch.State, + function: string + begin + Inc(Self.FProcExecutionCount); + Result := CExpectedResult; + end + ); Assert.AreEqual(0, FProcExecutionCount, 'AProc should not have executed yet.'); Assert.IsFalse(LFuture.Done.IsSet, 'Future should not be done yet.'); @@ -168,15 +168,15 @@ begin LInitStateAsState := TState.Null; // Immediate execution LFuture := - TMycGateFuncFuture.Create( - LLocalTaskFactory, - LInitStateAsState, - function: Integer - begin - Inc(Self.FProcExecutionCount); - raise Exception.Create(CExceptionMessage); - end - ); + TMycGateFuncFuture.Create( + LLocalTaskFactory, + LInitStateAsState, + function: Integer + begin + Inc(Self.FProcExecutionCount); + raise Exception.Create(CExceptionMessage); + end + ); try LLocalTaskFactory.WaitFor(LFuture.Done); @@ -225,15 +225,15 @@ begin LInitLatch := TLatch.Construct(1); LFuture := - TMycGateFuncFuture.Create( - FTaskFactory, - LInitLatch.State, - function: Integer - begin - Inc(Self.FProcExecutionCount); - Result := 123; - end - ); + TMycGateFuncFuture.Create( + FTaskFactory, + LInitLatch.State, + function: Integer + begin + Inc(Self.FProcExecutionCount); + Result := 123; + end + ); Assert.IsFalse(LFuture.Done.IsSet, 'Future should not be marked as done initially.'); @@ -264,44 +264,44 @@ begin // Create MainFuture, AInitState is the GateLatch's state. LMainFuture := - TMycGateFuncFuture.Create( - FTaskFactory, - LGateLatch.State, - function: string - begin - Inc(Self.FProcExecutionCount, 10); // Indicate MainFuture's proc ran - Result := CMainFutureResult; - end - ); + TMycGateFuncFuture.Create( + FTaskFactory, + LGateLatch.State, + function: string + begin + Inc(Self.FProcExecutionCount, 10); // Indicate MainFuture's proc ran + Result := CMainFutureResult; + end + ); // Setup Prerequisite Futures // PrerequisiteFuture1 LInitStateP1 := TLatch.Construct(1); // Controllable init state for PF1 LPrerequisiteFuture1 := - TMycGateFuncFuture.Create( - FTaskFactory, - LInitStateP1.State, - function: Integer - begin - Inc(Self.FProcExecutionCount, 1); // PF1 ran - Result := 1; - end - ); + TMycGateFuncFuture.Create( + FTaskFactory, + LInitStateP1.State, + function: Integer + begin + Inc(Self.FProcExecutionCount, 1); // PF1 ran + Result := 1; + end + ); LSub1 := TLatchNotifierSubscriber.Create(LGateLatch); Subscriptions[1] := LPrerequisiteFuture1.Done.Subscribe(LSub1); // Subscribe to PF1's completion [cite: 56, 188, 306] // PrerequisiteFuture2 LInitStateP2 := TLatch.Construct(1); // Controllable init state for PF2 LPrerequisiteFuture2 := - TMycGateFuncFuture.Create( - FTaskFactory, - LInitStateP2.State, - function: Integer - begin - Inc(Self.FProcExecutionCount, 1); // PF2 ran - Result := 2; - end - ); + TMycGateFuncFuture.Create( + FTaskFactory, + LInitStateP2.State, + function: Integer + begin + Inc(Self.FProcExecutionCount, 1); // PF2 ran + Result := 2; + end + ); LSub2 := TLatchNotifierSubscriber.Create(LGateLatch); Subscriptions[2] := LPrerequisiteFuture2.Done.Subscribe(LSub2); // Subscribe to PF2's completion @@ -347,29 +347,29 @@ begin // Future A LFutureA := - TMycGateFuncFuture.Create( - FTaskFactory, - LTriggerLatch.State, - function: Integer - begin - LFlagFutureARan := True; - System.SyncObjs.TInterlocked.Increment(Self.FSharedCounter); // Thread-safe increment - Result := 100; - end - ); + TMycGateFuncFuture.Create( + FTaskFactory, + LTriggerLatch.State, + function: Integer + begin + LFlagFutureARan := True; + System.SyncObjs.TInterlocked.Increment(Self.FSharedCounter); // Thread-safe increment + Result := 100; + end + ); // Future B LFutureB := - TMycGateFuncFuture.Create( - FTaskFactory, - LTriggerLatch.State, - function: Integer - begin - LFlagFutureBRan := True; - System.SyncObjs.TInterlocked.Increment(Self.FSharedCounter); // Thread-safe increment - Result := 200; - end - ); + TMycGateFuncFuture.Create( + FTaskFactory, + LTriggerLatch.State, + function: Integer + begin + LFlagFutureBRan := True; + System.SyncObjs.TInterlocked.Increment(Self.FSharedCounter); // Thread-safe increment + Result := 200; + end + ); Assert.IsFalse(LFutureA.Done.IsSet, 'FutureA should not be done yet.'); Assert.IsFalse(LFutureB.Done.IsSet, 'FutureB should not be done yet.'); diff --git a/Test/TestFutures.pas b/Test/TestFutures.pas index 528379a..89eeb65 100644 --- a/Test/TestFutures.pas +++ b/Test/TestFutures.pas @@ -75,13 +75,13 @@ var begin // Test construction with a simple function that returns an Integer fut := - TFuture.Construct( // [cite: 146] - function: Integer - begin - TThread.Sleep(20); // Simulate some background work - Result := 42; - end - ); + TFuture.Construct( // [cite: 146] + function: Integer + begin + TThread.Sleep(20); // Simulate some background work + Result := 42; + end + ); Assert.IsNotNull(fut.Done, 'Future.Done property should not be nil after construction.'); // Static string [cite: 148] fut.WaitFor(); // Wait for the future to complete its execution [cite: 148] @@ -99,14 +99,14 @@ var begin // Test construction with a nil gate, which should execute the task immediately fut := - TFuture.Construct( - nil, // Explicitly providing a nil gate [cite: 147] - function: Integer - begin - TThread.Sleep(20); // Simulate work - Result := 43; - end - ); + TFuture.Construct( + nil, // Explicitly providing a nil gate [cite: 147] + function: Integer + begin + TThread.Sleep(20); // Simulate work + Result := 43; + end + ); Assert.IsNotNull(fut.Done, 'Future.Done should not be nil when constructed with a nil gate.'); // Static string [cite: 148] fut.WaitFor(); // [cite: 148] @@ -128,13 +128,13 @@ begin // Construct a future with a gate that is already set fut := - TFuture.Construct( - presetGate, // [cite: 147] - function: Integer - begin - Result := 44; // This should execute quickly - end - ); + TFuture.Construct( + presetGate, // [cite: 147] + function: Integer + begin + Result := 44; // This should execute quickly + end + ); Assert.IsNotNull(fut.Done, 'Future.Done should not be nil for preset gate construct.'); // Static string [cite: 148] fut.WaitFor(); // [cite: 148] @@ -157,10 +157,10 @@ begin // Construct a future with a gate that is not yet set fut := - TFuture.Construct( - delayedGate.State, // Get the IMycState interface from the latch [cite: 147, 59] - function: Integer begin Result := 45; end - ); + TFuture.Construct( + delayedGate.State, // Get the IMycState interface from the latch [cite: 147, 59] + function: Integer begin Result := 45; end + ); Assert.IsNotNull(fut.Done, 'Future.Done should not be nil for delayed gate construct.'); // Static string [cite: 148] @@ -189,23 +189,23 @@ var begin // Create an initial future fut1 := - TFuture.Construct( // [cite: 146] - function: Integer - begin - TThread.Sleep(20); // Simulate work - Result := 100; - end - ); + TFuture.Construct( // [cite: 146] + function: Integer + begin + TThread.Sleep(20); // Simulate work + Result := 100; + end + ); // Chain a second future that depends on the result of the first fut2 := - fut1.Chain( // [cite: 147] - function(Input: Integer): string // This function receives the result of fut1 - begin - TThread.Sleep(20); // Simulate further work - Result := 'Value: ' + Input.ToString; // Use ToString for converting Integer to String - end - ); + fut1.Chain( // [cite: 147] + function(Input: Integer): string // This function receives the result of fut1 + begin + TThread.Sleep(20); // Simulate further work + Result := 'Value: ' + Input.ToString; // Use ToString for converting Integer to String + end + ); Assert.IsNotNull(fut2.Done, 'Chained Future.Done should not be nil.'); // Static string [cite: 148] fut2.WaitFor(); // Wait for the chained future to complete [cite: 148] @@ -229,16 +229,16 @@ begin // First future depends on the delayedGate fut1 := - TFuture.Construct( - delayedGate.State, // [cite: 147, 59] - function: Integer begin Result := 200; end - ); + TFuture.Construct( + delayedGate.State, // [cite: 147, 59] + function: Integer begin Result := 200; end + ); // Second future is chained to the first fut2 := - fut1.Chain( // [cite: 147] - function(Input: Integer): string begin Result := 'ChainVal: ' + Input.ToString; end - ); + fut1.Chain( // [cite: 147] + function(Input: Integer): string begin Result := 'ChainVal: ' + Input.ToString; end + ); Assert.IsNotNull(fut2.Done, 'Chained (with gate) Future.Done should not be nil.'); // Static string [cite: 148] @@ -268,33 +268,33 @@ var begin // Initial future A futA := - TFuture.Construct( // [cite: 146] - function: Integer - begin - TThread.Sleep(10); - Result := 10; - end - ); + TFuture.Construct( // [cite: 146] + function: Integer + begin + TThread.Sleep(10); + Result := 10; + end + ); // Future B, chained from A futB := - futA.Chain( // [cite: 147] - function(InputA: Integer): Real - begin - TThread.Sleep(10); - Result := InputA * 2.5; // Calculation: 10 * 2.5 = 25.0 - end - ); + futA.Chain( // [cite: 147] + function(InputA: Integer): Real + begin + TThread.Sleep(10); + Result := InputA * 2.5; // Calculation: 10 * 2.5 = 25.0 + end + ); // Future C, chained from B futC := - futB.Chain( // [cite: 147] - function(InputB: Real): string - begin - TThread.Sleep(10); - Result := 'Final: ' + FloatToStr(InputB); // Convert Real to String - end - ); + futB.Chain( // [cite: 147] + function(InputB: Real): string + begin + TThread.Sleep(10); + Result := 'Final: ' + FloatToStr(InputB); // Convert Real to String + end + ); Assert.IsNotNull(futC.Done, 'Multi-chained Future.Done should not be nil.'); // Static string [cite: 148] futC.WaitFor(); // Wait for the last future in the chain [cite: 148] @@ -318,13 +318,13 @@ var begin // Parametric test for Future fut := - TFuture.Construct( // [cite: 146] - function: string - begin - TThread.Sleep(10); - Result := ParamValue; // Use the parameter in the future's function - end - ); + TFuture.Construct( // [cite: 146] + function: string + begin + TThread.Sleep(10); + Result := ParamValue; // Use the parameter in the future's function + end + ); Assert.IsNotNull(fut.Done, 'Parametric Future.Done should not be nil.'); // Static string [cite: 148] fut.WaitFor(); // [cite: 148] @@ -354,22 +354,22 @@ begin begin // Capture loop variable for use in anonymous method Futures[i] := - TFuture.Construct( // [cite: 146] - ( - function(captureIndex: Integer): TFunc - begin - Result := - function: Integer - var - delay: Integer; - begin - delay := 10 + Random(40); // Random delay between 10ms and 49ms - TThread.Sleep(delay); - Result := captureIndex; // Return the captured index - end; - end - )(i) - ); + TFuture.Construct( // [cite: 146] + ( + function(captureIndex: Integer): TFunc + begin + Result := + function: Integer + var + delay: Integer; + begin + delay := 10 + Random(40); // Random delay between 10ms and 49ms + TThread.Sleep(delay); + Result := captureIndex; // Return the captured index + end; + end + )(i) + ); doneStates[i] := Futures[i].Done; // [cite: 148] end; @@ -378,13 +378,13 @@ begin // Create a master future that waits on the combined State.All state masterFuture := - TFuture.Construct( - combinedStateAll, // [cite: 147] - function: Boolean - begin - Result := True; // This function executes when combinedStateAll is set - end - ); + TFuture.Construct( + combinedStateAll, // [cite: 147] + function: Boolean + begin + Result := True; // This function executes when combinedStateAll is set + end + ); masterFuture.WaitFor(); // Wait for all futures to complete via the master future [cite: 148] Assert.IsTrue(combinedStateAll.IsSet, 'Combined State.All should be set after masterFuture.WaitFor().'); // Static string [cite: 55] @@ -424,25 +424,25 @@ begin begin // Capture loop variable Futures[i] := - TFuture.Construct( // [cite: 146] - ( - function(captureIndex: Integer): TFunc - begin - Result := - function: Integer - var - delay: Integer; - begin - if captureIndex = QuickFutureIndex then - delay := 5 // Short delay for the 'quick' future - else - delay := 50 + Random(100); // Longer random delay (50-149ms) for others - TThread.Sleep(delay); - Result := captureIndex; - end; - end - )(i) - ); + TFuture.Construct( // [cite: 146] + ( + function(captureIndex: Integer): TFunc + begin + Result := + function: Integer + var + delay: Integer; + begin + if captureIndex = QuickFutureIndex then + delay := 5 // Short delay for the 'quick' future + else + delay := 50 + Random(100); // Longer random delay (50-149ms) for others + TThread.Sleep(delay); + Result := captureIndex; + end; + end + )(i) + ); doneStates[i] := Futures[i].Done; // [cite: 148] end; @@ -451,13 +451,13 @@ begin // Create a master future that waits on the combined State.Any state masterFuture := - TFuture.Construct( - combinedStateAny, // [cite: 147] - function: Boolean - begin - Result := True; // This function executes when combinedStateAny is set - end - ); + TFuture.Construct( + combinedStateAny, // [cite: 147] + function: Boolean + begin + Result := True; // This function executes when combinedStateAny is set + end + ); masterFuture.WaitFor(); // Wait for at least one future to complete [cite: 148] Assert.IsTrue(combinedStateAny.IsSet, 'Combined State.Any should be set after masterFuture.WaitFor().'); // Static string [cite: 55] @@ -492,20 +492,20 @@ var begin // Create an outer future that, when resolved, produces another (inner) future. outerFuture := - TFuture>.Construct( // [cite: 146] - function: TFuture // This lambda returns a Future - begin - TThread.Sleep(10); // Simulate work for the outer future to produce the inner one - Result := - TFuture.Construct( // [cite: 146] - function: Integer - begin - TThread.Sleep(10); // Simulate work for the inner future - Result := 123; // The final value - end - ); - end + TFuture>.Construct( // [cite: 146] + function: TFuture // This lambda returns a Future + begin + TThread.Sleep(10); // Simulate work for the outer future to produce the inner one + Result := + TFuture.Construct( // [cite: 146] + function: Integer + begin + TThread.Sleep(10); // Simulate work for the inner future + Result := 123; // The final value + end ); + end + ); Assert.IsNotNull(outerFuture.Done, 'Outer future''s Done state should not be nil.'); // Static string [cite: 148] outerFuture.WaitFor(); // Wait for the outer future to complete and yield the inner future [cite: 148] @@ -531,31 +531,31 @@ var begin // Create an initial future initialFuture := - TFuture.Construct( // [cite: 146] - function: Integer - begin - TThread.Sleep(10); - Result := 77; - end - ); + TFuture.Construct( // [cite: 146] + function: Integer + begin + TThread.Sleep(10); + Result := 77; + end + ); // Chain it with a function that itself returns a new Future outerChainedFuture := - initialFuture.Chain>( // [cite: 147] - function(Input: Integer): TFuture // This lambda returns a Future - begin - TThread.Sleep(10); // Simulate work in the chain function - // Input is the result of initialFuture (77) - Result := - TFuture.Construct( // [cite: 146] - function: string - begin - TThread.Sleep(10); // Simulate work for the inner-most future - Result := 'Value: ' + Input.ToString; // Input is captured (77) - end - ); - end + initialFuture.Chain>( // [cite: 147] + function(Input: Integer): TFuture // This lambda returns a Future + begin + TThread.Sleep(10); // Simulate work in the chain function + // Input is the result of initialFuture (77) + Result := + TFuture.Construct( // [cite: 146] + function: string + begin + TThread.Sleep(10); // Simulate work for the inner-most future + Result := 'Value: ' + Input.ToString; // Input is captured (77) + end ); + end + ); Assert.IsNotNull(outerChainedFuture.Done, 'Outer chained future''s Done state should not be nil.'); // Static string [cite: 148] outerChainedFuture.WaitFor(); // Wait for initialFuture to complete AND the chain function to execute [cite: 148] diff --git a/Test/TestNotifier.pas b/Test/TestNotifier.pas index 8d307b9..3bf48c0 100644 --- a/Test/TestNotifier.pas +++ b/Test/TestNotifier.pas @@ -106,11 +106,11 @@ var begin predicateCallCount := 0; dummyPredicate := - function(Item: IMyTestInterface): Boolean - begin - Inc(predicateCallCount); - Result := True; - end; + function(Item: IMyTestInterface): Boolean + begin + Inc(predicateCallCount); + Result := True; + end; FNotifier.Lock; Assert.IsTrue(FNotifier.IsLocked, 'Notifier should be locked.'); @@ -209,12 +209,12 @@ begin itemsProcessedCount := 0; FNotifier.Notify( - function(Item: IMyTestInterface): Boolean - begin - Inc(itemsProcessedCount); - TMyTestReceiver(Item).Foo; - Result := Item.GetValue <> 20; // Remove item with value 20 - end + function(Item: IMyTestInterface): Boolean + begin + Inc(itemsProcessedCount); + TMyTestReceiver(Item).Foo; + Result := Item.GetValue <> 20; // Remove item with value 20 + end ); Assert.AreEqual(3, itemsProcessedCount, 'Notify predicate called for all 3 items.'); @@ -229,12 +229,12 @@ begin itemsProcessedCount := 0; FNotifier.Notify( - function(Item: IMyTestInterface): Boolean - begin - Inc(itemsProcessedCount); - TMyTestReceiver(Item).Foo; - Result := True; // Keep remaining - end + function(Item: IMyTestInterface): Boolean + begin + Inc(itemsProcessedCount); + TMyTestReceiver(Item).Foo; + Result := True; // Keep remaining + end ); Assert.AreEqual(2, itemsProcessedCount, 'Notify predicate called for 2 remaining items.'); Assert.AreEqual(1, rcv1.CallCount, 'Receiver1 called again.'); diff --git a/Test/TestNotifier_ChaosStress.pas b/Test/TestNotifier_ChaosStress.pas index 0b0155c..3baf431 100644 --- a/Test/TestNotifier_ChaosStress.pas +++ b/Test/TestNotifier_ChaosStress.pas @@ -213,11 +213,11 @@ begin // if not FOwnerFixture.FNotifier.IsFinalized then // Don't notify if finalized begin FOwnerFixture.FNotifier.Notify( - function(Item: IMyStressTestInterface): Boolean - begin - Item.Foo(FThreadID); // Pass ThreadID as notification type for context - Result := True; // Keep item - end + function(Item: IMyStressTestInterface): Boolean + begin + Item.Foo(FThreadID); // Pass ThreadID as notification type for context + Result := True; // Keep item + end ); end; finally @@ -359,11 +359,11 @@ begin // if not FNotifier.IsFinalized then begin FNotifier.Notify( - function(Item: IMyStressTestInterface): Boolean - begin - actualLiveReceiversInNotifier.Add(Item); - Result := True; - end + function(Item: IMyStressTestInterface): Boolean + begin + actualLiveReceiversInNotifier.Add(Item); + Result := True; + end ); end; finally @@ -373,12 +373,12 @@ begin // 2. Compare overall counts Assert.AreEqual( - totalExpectedLiveByThreadsAtEnd, - actualLiveInNotifierAtEnd, - Format( - 'Mismatch in live item count at end. Threads expected %d, Notifier has %d.', - [totalExpectedLiveByThreadsAtEnd, actualLiveInNotifierAtEnd] - ) + totalExpectedLiveByThreadsAtEnd, + actualLiveInNotifierAtEnd, + Format( + 'Mismatch in live item count at end. Threads expected %d, Notifier has %d.', + [totalExpectedLiveByThreadsAtEnd, actualLiveInNotifierAtEnd] + ) ); // 3. Detailed check: Iterate all receivers ever created. @@ -400,27 +400,22 @@ begin end; Assert.AreEqual( - receiver.ExpectedToBeAdvised, - found, - Format( - 'Receiver ID %d: Thread expected it to be advised=%d, but its presence in Notifier is %d. NotificationCount=%d', - [ - receiver.GetInstanceID, - Integer(receiver.ExpectedToBeAdvised), - Integer(found), - receiver.GetNotificationCount - ] - ) + receiver.ExpectedToBeAdvised, + found, + Format( + 'Receiver ID %d: Thread expected it to be advised=%d, but its presence in Notifier is %d. NotificationCount=%d', + [receiver.GetInstanceID, Integer(receiver.ExpectedToBeAdvised), Integer(found), receiver.GetNotificationCount] + ) ); // Further checks on NotificationCount could be added if specific notification patterns were expected. // For this chaos test, ensuring count is non-negative and consistent with advised state is a good start. Assert.IsTrue( - receiver.GetNotificationCount >= 0, - Format( - 'Receiver ID %d has non-positive notification count: %d', - [receiver.GetInstanceID, receiver.GetNotificationCount] - ) + receiver.GetNotificationCount >= 0, + Format( + 'Receiver ID %d has non-positive notification count: %d', + [receiver.GetInstanceID, receiver.GetNotificationCount] + ) ); if found and (receiver.GetNotificationCount = 0) then begin diff --git a/Test/TestNotifier_Threading.pas b/Test/TestNotifier_Threading.pas index 2fc3dd7..d5057b0 100644 --- a/Test/TestNotifier_Threading.pas +++ b/Test/TestNotifier_Threading.pas @@ -256,23 +256,23 @@ begin FNotifier.Lock; try FNotifier.Notify( - function(Item: IMyTestInterface): Boolean - begin - Inc(currentNotifyCount); - Result := True; // Keep item in the Notifier - end + function(Item: IMyTestInterface): Boolean + begin + Inc(currentNotifyCount); + Result := True; // Keep item in the Notifier + end ); finally FNotifier.Release; end; Assert.AreEqual( - totalAdvisedExpected, - currentNotifyCount, - 'The number of items in the Notifier after all Advise operations (' - + currentNotifyCount.ToString - + ') does not match the expected number (' - + totalAdvisedExpected.ToString - + ').' + totalAdvisedExpected, + currentNotifyCount, + 'The number of items in the Notifier after all Advise operations (' + + currentNotifyCount.ToString + + ') does not match the expected number (' + + totalAdvisedExpected.ToString + + ').' ); // 2. Execute UnadviseAll @@ -288,19 +288,19 @@ begin FNotifier.Lock; try FNotifier.Notify( - function(Item: IMyTestInterface): Boolean - begin - Inc(currentNotifyCount); - Result := True; - end + function(Item: IMyTestInterface): Boolean + begin + Inc(currentNotifyCount); + Result := True; + end ); finally FNotifier.Release; end; Assert.AreEqual( - 0, - currentNotifyCount, - 'The Notifier should be empty after UnadviseAll, but still contains ' + currentNotifyCount.ToString + ' items.' + 0, + currentNotifyCount, + 'The Notifier should be empty after UnadviseAll, but still contains ' + currentNotifyCount.ToString + ' items.' ); end; diff --git a/Test/TestSignals_Latch.pas b/Test/TestSignals_Latch.pas index a892a81..1d9e831 100644 --- a/Test/TestSignals_Latch.pas +++ b/Test/TestSignals_Latch.pas @@ -129,9 +129,9 @@ begin end; procedure TTestMycLatch.TestNotify_ReturnValueAndState_AfterOneNotify( - InitialCount: Integer; - ExpectedReturn: Boolean; - ExpectedIsSet: Boolean + InitialCount: Integer; + ExpectedReturn: Boolean; + ExpectedIsSet: Boolean ); var latch: IMycLatch; @@ -141,14 +141,14 @@ begin returnedValueFromNotify := latch.Notify; // This is IMycLatch (as IMycSubscriber).Notify Assert.AreEqual( - ExpectedReturn, - returnedValueFromNotify, - 'Unexpected return value from Latch.Notify call for initial count ' + IntToStr(InitialCount) + '.' + ExpectedReturn, + returnedValueFromNotify, + 'Unexpected return value from Latch.Notify call for initial count ' + IntToStr(InitialCount) + '.' ); Assert.AreEqual( - ExpectedIsSet, - latch.State.IsSet, - 'Unexpected IsSet state after Latch.Notify call for initial count ' + IntToStr(InitialCount) + '.' + ExpectedIsSet, + latch.State.IsSet, + 'Unexpected IsSet state after Latch.Notify call for initial count ' + IntToStr(InitialCount) + '.' ); end; @@ -504,11 +504,8 @@ begin Assert.AreEqual(1, mockSubInitial.NotifyCount, 'Initial subscriber (unadvised) not notified again.'); // mockSubNew should also not be notified again, as it was only notified immediately // and not persistently added to FSubscribers for subsequent Latch.Notify calls. - Assert.AreEqual( - 1, - mockSubNew.NotifyCount, - 'New subscriber (notified once on subscribe) not notified by subsequent Latch.Notify calls.' - ); + Assert + .AreEqual(1, mockSubNew.NotifyCount, 'New subscriber (notified once on subscribe) not notified by subsequent Latch.Notify calls.'); end; initialization diff --git a/Test/TestTasks.pas b/Test/TestTasks.pas index 8366bda..54bf882 100644 --- a/Test/TestTasks.pas +++ b/Test/TestTasks.pas @@ -92,13 +92,13 @@ begin jobCompletedLatch := TLatch.Construct(1); // Changed from Signals.CreateLatch FFactory - .Run( - procedure - begin - jobExecuted := True; - jobCompletedLatch.Notify; - end) - .Notify; + .Run( + procedure + begin + jobExecuted := True; + jobCompletedLatch.Notify; + end) + .Notify; FFactory.WaitFor(jobCompletedLatch.State); Assert.IsTrue(jobExecuted, 'Immediate job should have executed'); @@ -114,13 +114,13 @@ begin jobCompletedLatch := TLatch.Construct(1); // Changed from Signals.CreateLatch subscriber := - FFactory.Run( - procedure - begin - jobExecuted := True; - jobCompletedLatch.Notify; - end - ); + FFactory.Run( + procedure + begin + jobExecuted := True; + jobCompletedLatch.Notify; + end + ); Assert.IsNotNull(subscriber, 'Run should return a subscriber for delayed job'); // Use TMycLatch.Null for comparison @@ -162,12 +162,12 @@ begin Assert.IsFalse(FFactory.InWorkerThread, 'Test method itself should not be in a factory worker thread'); FFactory.EnqueueJob( - procedure - begin - inMainThreadInJob := FFactory.InMainThread; - inWorkerThreadInJob := FFactory.InWorkerThread; - jobDoneLatch.Notify; - end + procedure + begin + inMainThreadInJob := FFactory.InMainThread; + inWorkerThreadInJob := FFactory.InWorkerThread; + jobDoneLatch.Notify; + end ); FFactory.WaitFor(jobDoneLatch.State); @@ -183,20 +183,20 @@ begin jobDoneLatch := TLatch.Construct(1); FFactory.EnqueueJob( - procedure - begin - try - raise TMycTaskFactory.ETaskException.Create('Test Exception From Job'); - finally - jobDoneLatch.Notify; - end; - end + procedure + begin + try + raise TMycTaskFactory.ETaskException.Create('Test Exception From Job'); + finally + jobDoneLatch.Notify; + end; + end ); Assert.WillRaise( - procedure begin FFactory.WaitFor(jobDoneLatch.State); end, - TMycTaskFactory.ETaskException, - 'WaitFor should re-raise the exception from the job' + procedure begin FFactory.WaitFor(jobDoneLatch.State); end, + TMycTaskFactory.ETaskException, + 'WaitFor should re-raise the exception from the job' ); var noExceptionRaised: Boolean := True; @@ -213,9 +213,9 @@ begin FFactory.Teardown; Assert.WillRaise( - procedure begin FFactory.EnqueueJob(procedure begin end); end, - TMycTaskFactory.ETaskException, - 'Running a job on a torn-down factory should raise ETaskException' + procedure begin FFactory.EnqueueJob(procedure begin end); end, + TMycTaskFactory.ETaskException, + 'Running a job on a torn-down factory should raise ETaskException' ); end; @@ -230,18 +230,18 @@ begin dummyStateToWaitFor := TLatch.Construct(1).State; FFactory.EnqueueJob( - procedure - begin - try - FFactory.WaitFor(dummyStateToWaitFor); - except - on E: TMycTaskFactory.ETaskException do - begin - exceptionCaughtInJob := True; - end; - end; - jobDoneLatch.Notify; - end + procedure + begin + try + FFactory.WaitFor(dummyStateToWaitFor); + except + on E: TMycTaskFactory.ETaskException do + begin + exceptionCaughtInJob := True; + end; + end; + jobDoneLatch.Notify; + end ); FFactory.WaitFor(jobDoneLatch.State);