diff --git a/README.md b/README.md index 0dcc6873..fdb3ba12 100644 --- a/README.md +++ b/README.md @@ -63,6 +63,7 @@ ### ⛓️ Flow - [`Flow`](modules/Language/XML/Flow.mpp) - Declarative data transformation and file orchestration with staging, watching, and parallel chunking +- [`FlowRunner`](modules/Language/XML/FlowRunner.mpp) - Orchestrator discovering, indexing, and executing XML flows in parallel - [`Pipeline`](modules/Language/XML/Pipeline.mpp) - Lazy C++20 range pipeline compiled from XML tags - [`Tags`](modules/Language/XML/Tags.mpp) - Standard pipeline tags (``, ``, ``, ``, ``, ``, ``, ``, ``, ``) @@ -105,6 +106,7 @@ - [`Multiton`](modules/Pattern/Multiton.mpp) - Generic Multiton implementation based on the Meyers Singleton ### 📶 Ranges +- [`Drain`](modules/Ranges/Drain.mpp) - Consuming range adaptor draining elements from a range - [`Expected`](modules/Ranges/Expected.mpp) - Range adaptors for `std::expected` streams - [`Inspect`](modules/Ranges/Inspect.mpp) - Range element inspection for side-effects and logging - [`Parallel`](modules/Ranges/Parallel.mpp) - Multithreaded range pipelining diff --git a/modules/Language/Language.mpp b/modules/Language/Language.mpp index 209f5a1a..8cf85925 100644 --- a/modules/Language/Language.mpp +++ b/modules/Language/Language.mpp @@ -15,6 +15,7 @@ export import CppUtils.Language.Xml; export import CppUtils.Language.Xml.Pipeline; export import CppUtils.Language.Xml.Tags; export import CppUtils.Language.Xml.Flow; +export import CppUtils.Language.Xml.FlowRunner; export import CppUtils.Language.Ini; export import CppUtils.Language.Json; export import CppUtils.Language.Parameters; diff --git a/modules/Language/XML/Flow.mpp b/modules/Language/XML/Flow.mpp index 76a5ac34..0ff73aa0 100644 --- a/modules/Language/XML/Flow.mpp +++ b/modules/Language/XML/Flow.mpp @@ -9,6 +9,7 @@ import CppUtils.FileSystem.Watcher; import CppUtils.FileSystem.FileStaging; import CppUtils.Thread.ThreadPool; import CppUtils.Thread.UniqueLocker; +import CppUtils.Ranges.Drain; import CppUtils.Ranges.Parallel; import CppUtils.Language.Xml; import CppUtils.Language.Xml.Pipeline; @@ -19,14 +20,29 @@ import CppUtils.Logger; export namespace CppUtils::Language::Xml { + class Flow + { + public: + struct Report final + { + std::size_t processedFilesCount = 0; + }; + + virtual ~Flow() = default; + + virtual auto execute() -> Report = 0; + virtual auto startWatching() -> void = 0; + virtual auto stopWatching() -> void = 0; + }; + template - class Flow final + class TypedFlow final: public Flow { public: - using TagHandler = std::function& node, Flow& flow)>; - using FormatProcessor = std::function(const std::filesystem::path& inputFilePath, const std::filesystem::path& outputFilePath)>; + using TagHandler = std::function& node, TypedFlow& flow)>; + using FormatProcessor = std::function; - explicit Flow(Thread::ThreadPool& threadPool): m_threadPool{std::ref(threadPool)} + explicit TypedFlow(Thread::ThreadPool& threadPool): m_threadPool{std::ref(threadPool)} {} auto registerTagHandler(Type::Token name, TagHandler handler) -> void @@ -69,17 +85,18 @@ export namespace CppUtils::Language::Xml dispatchNode(operator""_xml(std::ranges::data(xmlSource), std::ranges::size(xmlSource))); } - auto execute() -> std::vector + auto execute() -> Report override { + auto report = Report{}; + if (m_watch and m_sourceDirectory.has_value()) { startWatching(); - return {}; + return report; } if (m_sourceDirectory.has_value()) { - auto processedObjects = std::vector{}; if (m_outputDirectory.has_value()) std::filesystem::create_directories(m_outputDirectory.value()); @@ -87,19 +104,22 @@ export namespace CppUtils::Language::Xml for (const auto& entry : std::filesystem::directory_iterator(m_sourceDirectory.value())) if (entry.is_regular_file() and matchesPattern(entry.path())) { - auto fileResults = processFileWithStaging(entry.path(), computeOutputFile(entry.path())); - processedObjects.insert(std::ranges::cend(processedObjects), std::ranges::begin(fileResults), std::ranges::end(fileResults)); + processFileWithStaging(entry.path(), computeOutputFile(entry.path())); + ++report.processedFilesCount; } - return processedObjects; + return report; } if (m_sourceFile.has_value()) - return processFileWithStaging(m_sourceFile.value(), computeOutputFile(m_sourceFile.value())); + { + processFileWithStaging(m_sourceFile.value(), computeOutputFile(m_sourceFile.value())); + ++report.processedFilesCount; + } - return {}; + return report; } - auto startWatching() -> void + auto startWatching() -> void override { if (not m_sourceDirectory.has_value() or m_watcher) return; @@ -115,7 +135,7 @@ export namespace CppUtils::Language::Xml m_watcher->watch(m_sourceDirectory.value()); } - auto stopWatching() -> void + auto stopWatching() -> void override { if (not m_watcher) return; @@ -182,7 +202,7 @@ export namespace CppUtils::Language::Xml return filePipeline; } - auto processCsvFile(const std::filesystem::path& inputFilePath, const std::filesystem::path& outputFilePath) -> std::vector + auto processCsvFile(const std::filesystem::path& inputFilePath, const std::filesystem::path& outputFilePath) -> void { const auto effectiveChunkSize = m_chunkSize > 0 ? m_chunkSize : 100uz; @@ -196,30 +216,32 @@ export namespace CppUtils::Language::Xml }); if (not std::ranges::empty(outputFilePath)) - { processedObjects | CppUtils::Language::CSV::toCSV(m_outputSeparator) | CppUtils::FileSystem::writeLines(outputFilePath.string()); - return {}; - } - - return processedObjects | std::ranges::to>(); + else + processedObjects | CppUtils::Ranges::drain; } - auto processFile(const std::filesystem::path& inputFilePath, const std::filesystem::path& outputFilePath) -> std::vector + auto processFile(const std::filesystem::path& inputFilePath, const std::filesystem::path& outputFilePath) -> void { const auto formatToken = Type::hash(m_sourceFormat); if (const auto iterator = m_formatRegistry.find(formatToken); iterator != std::ranges::end(m_formatRegistry)) - return iterator->second(inputFilePath, outputFilePath); + { + iterator->second(inputFilePath, outputFilePath); + return; + } Logger<"Flow">::template print<"error">("Unsupported source format: {}", m_sourceFormat); - return {}; } - auto processFileWithStaging(const std::filesystem::path& inputFilePath, const std::filesystem::path& outputFilePath) -> std::vector + auto processFileWithStaging(const std::filesystem::path& inputFilePath, const std::filesystem::path& outputFilePath) -> void { if (not m_stagingDirectory.has_value()) - return processFile(inputFilePath, outputFilePath); + { + processFile(inputFilePath, outputFilePath); + return; + } auto stageResult = CppUtils::FileSystem::FileStaging::stage( inputFilePath, @@ -229,23 +251,21 @@ export namespace CppUtils::Language::Xml if (not stageResult.has_value()) { Logger<"Flow">::template print<"error">("Failed to stage file {}: {}", inputFilePath.string(), stageResult.error().message()); - return {}; + return; } auto& staging = stageResult.value(); try { - auto result = processFile(staging.stagedPath(), outputFilePath); + processFile(staging.stagedPath(), outputFilePath); if (auto completeResult = staging.complete(); not completeResult) Logger<"Flow">::template print<"error">("Failed to complete staging for {}: {}", inputFilePath.string(), completeResult.error().message()); - return result; } catch (const std::exception& exception) { Logger<"Flow">::template print<"error">("Error processing file {}: {}", inputFilePath.string(), exception.what()); if (auto abortResult = staging.abort(); not abortResult) Logger<"Flow">::template print<"error">("Failed to abort staging for {}: {}", inputFilePath.string(), abortResult.error().message()); - return {}; } } @@ -373,30 +393,30 @@ export namespace CppUtils::Language::Xml }; template - [[nodiscard]] inline auto buildFlow(const Container::Tree::VariantNode& node, Thread::ThreadPool& threadPool) -> Flow + [[nodiscard]] inline auto buildFlow(const Container::Tree::VariantNode& node, Thread::ThreadPool& threadPool) -> TypedFlow { - auto flow = Flow{threadPool}; + auto flow = TypedFlow{threadPool}; flow.template registerStandardTags(); flow.dispatchNode(node); return flow; } template - [[nodiscard]] inline auto buildFlow(std::string_view xmlSource, Thread::ThreadPool& threadPool) -> Flow + [[nodiscard]] inline auto buildFlow(std::string_view xmlSource, Thread::ThreadPool& threadPool) -> TypedFlow { using namespace CppUtils::Language::Xml::Literals; return buildFlow(operator""_xml(std::ranges::data(xmlSource), std::ranges::size(xmlSource)), threadPool); } template - inline auto executeFlow(const Container::Tree::VariantNode& node, Thread::ThreadPool& threadPool) -> std::vector + inline auto executeFlow(const Container::Tree::VariantNode& node, Thread::ThreadPool& threadPool) -> void { - return buildFlow(node, threadPool).execute(); + buildFlow(node, threadPool).execute(); } template - inline auto executeFlow(std::string_view xmlSource, Thread::ThreadPool& threadPool) -> std::vector + inline auto executeFlow(std::string_view xmlSource, Thread::ThreadPool& threadPool) -> void { - return buildFlow(xmlSource, threadPool).execute(); + buildFlow(xmlSource, threadPool).execute(); } } diff --git a/modules/Language/XML/FlowRunner.mpp b/modules/Language/XML/FlowRunner.mpp new file mode 100644 index 00000000..1e5315e5 --- /dev/null +++ b/modules/Language/XML/FlowRunner.mpp @@ -0,0 +1,266 @@ +export module CppUtils.Language.Xml.FlowRunner; + +import std; +import CppUtils.Type.Concept; +import CppUtils.String; +import CppUtils.Container.Tree; +import CppUtils.FileSystem; +import CppUtils.Thread.ThreadPool; +import CppUtils.Logger; +import CppUtils.Chrono.Chronometer; +import CppUtils.Container.MultiKeyMap; +import CppUtils.Language.Xml; +import CppUtils.Language.Xml.Flow; + +export namespace CppUtils::Language::Xml +{ + class FlowRunner final + { + public: + struct Stats final + { + std::size_t processedFilesCount = 0; + std::chrono::steady_clock::duration elapsedDuration = std::chrono::steady_clock::duration::zero(); + }; + + using FlowNode = Container::Tree::VariantNode; + using FlowFactory = std::function(const FlowNode&)>; + + explicit FlowRunner(Thread::ThreadPool& threadPool, + std::filesystem::path flowsDirectory = "data/flows"): + m_threadPool{threadPool}, + m_flowsDirectory{std::move(flowsDirectory)} + {} + + FlowRunner(const FlowRunner&) = delete; + FlowRunner(FlowRunner&&) = delete; + auto operator=(const FlowRunner&) -> FlowRunner& = delete; + auto operator=(FlowRunner&&) -> FlowRunner& = delete; + + private: + template + [[nodiscard]] auto createFlow( + const FlowNode& flowNode, + const std::function&)>& configure = nullptr) -> std::unique_ptr + { + auto flow = std::make_unique>(m_threadPool); + if (configure) + configure(*flow); + else + flow->template registerStandardTags(); + flow->dispatchNode(flowNode); + return flow; + } + + public: + template + auto registerMapping( + std::string_view identifier, + std::function&)> configure = nullptr) -> void + { + m_factories[std::string{identifier}] = + [this, configure = std::move(configure)](const FlowNode& flowNode) { + return createFlow(flowNode, configure); + }; + } + + using FlowContainer = Container::MultiKeyMap>; + + private: + [[nodiscard]] auto findFlowNode(const FlowNode& rootNode) const -> const FlowNode& + { + using namespace String::Literals; + if (rootNode.exists("Flow"_token)) + return rootNode["Flow"_token]; + return rootNode; + } + + [[nodiscard]] auto extractFlowName(const FlowNode& flowNode, const std::filesystem::path& filePath) const -> std::string + { + using namespace String::Literals; + if (const auto flowName = Xml::extractAttribute(flowNode, "name"_token); flowName.has_value() and not std::ranges::empty(flowName.value())) + return std::move(flowName.value()); + return filePath.stem().string(); + } + + [[nodiscard]] auto resolveFactory(const FlowNode& flowNode, const std::filesystem::path& filePath) const -> const FlowFactory& + { + using namespace String::Literals; + + if (const auto explicitMapping = Xml::extractAttribute(flowNode, "mapping"_token)) + if (const auto iterator = m_factories.find(std::string{explicitMapping.value()}); iterator != std::ranges::end(m_factories)) + return iterator->second; + + if (const auto iterator = m_factories.find(""); iterator != std::ranges::end(m_factories)) + return iterator->second; + + throw std::runtime_error{std::format("No flow mapping factory found for config: {}", filePath.string())}; + } + + public: + auto loadFlows() -> void + { + for (const auto& entry : m_flows) + entry.value->stopWatching(); + m_flows.clear(); + + for (const auto& filePath : discoverFlowConfigs()) + try + { + const auto rootNode = Xml::parse(FileSystem::String::read(filePath)); + const auto& flowNode = findFlowNode(rootNode); + const auto flowName = extractFlowName(flowNode, filePath); + const auto& factory = resolveFactory(flowNode, filePath); + m_flows.emplace(filePath, flowName, std::shared_ptr{factory(flowNode)}); + } + catch (const std::exception& exception) + { + Logger<"Flow">::template print<"warning">( + "Failed to load flow from '{}': {}", filePath.string(), exception.what()); + } + } + + [[nodiscard]] auto flows() const -> const FlowContainer& + { + return m_flows; + } + + [[nodiscard]] auto flow(const Type::Exact auto& filePath) const -> std::shared_ptr + { + const auto iterator = m_flows.find(filePath); + return iterator != std::ranges::end(m_flows) ? iterator->value : nullptr; + } + + [[nodiscard]] auto flow(std::string_view flowName) const -> std::shared_ptr + { + const auto iterator = m_flows.template findAt<1>(flowName); + return iterator != std::ranges::end(m_flows) ? iterator->value : nullptr; + } + + [[nodiscard]] auto discoverFlowConfigs() const -> std::vector + { + auto configs = std::vector{}; + if (std::filesystem::exists(m_flowsDirectory)) + { + FileSystem::forFilesWithExtension( + m_flowsDirectory, ".xml", [&](const auto& filePath) { configs.push_back(filePath); }); + std::ranges::sort(configs); + } + return configs; + } + + auto execute() -> Stats + { + if (std::ranges::empty(m_flows)) + { + Logger<"Flow">::template print<"warning">( + "FlowRunner: 0 active flows found to execute (did you call loadFlows()?)"); + return Stats{}; + } + + auto futures = std::vector>{}; + futures.reserve(std::ranges::size(m_flows)); + + auto chronometer = Chrono::Chronometer{}; + for (const auto& entry : m_flows) + futures.push_back(std::async(std::launch::async, [flow = entry.value] { + return flow->execute(); + })); + + auto processedFilesCount = 0uz; + for (auto& future : futures) + { + const auto report = future.get(); + processedFilesCount += report.processedFilesCount; + } + chronometer.stop(); + + return Stats{ + .processedFilesCount = processedFilesCount, + .elapsedDuration = chronometer.elapsed()}; + } + + auto startWatching() -> void + { + if (not std::filesystem::exists(m_flowsDirectory)) + { + Logger<"Flow">::template print<"warning">( + "FlowRunner: Flows directory '{}' does not exist", m_flowsDirectory.string()); + return; + } + + if (std::ranges::empty(m_flows)) + loadFlows(); + + for (const auto& entry : m_flows) + entry.value->startWatching(); + + m_flowsWatcher = std::make_unique(); + m_flowsWatcher->onEvent([this](FileSystem::Event event, const std::filesystem::path& filePath) { + if (filePath.extension() != ".xml") + return; + + if (event & (FileSystem::Event::Deleted | FileSystem::Event::MovedFrom)) + { + if (m_flows.contains(filePath)) + { + if (auto existingFlow = flow(filePath)) + existingFlow->stopWatching(); + m_flows.erase(filePath); + Logger<"Flow">::template print<"info">( + "Workflow '{}' removed from active flows", filePath.filename().string()); + } + return; + } + + if (event & (FileSystem::Event::Created | FileSystem::Event::MovedTo | FileSystem::Event::Modified | FileSystem::Event::CloseWrite)) + try + { + const auto rootNode = Xml::parse(FileSystem::String::read(filePath)); + const auto& flowNode = findFlowNode(rootNode); + const auto& factory = resolveFactory(flowNode, filePath); + auto newFlow = std::shared_ptr{factory(flowNode)}; + + const auto isUpdate = m_flows.contains(filePath); + if (isUpdate) + { + if (auto existingFlow = flow(filePath)) + existingFlow->stopWatching(); + m_flows.erase(filePath); + } + + newFlow->startWatching(); + const auto flowName = extractFlowName(flowNode, filePath); + m_flows.emplace(filePath, flowName, std::move(newFlow)); + + if (isUpdate) + Logger<"Flow">::template print<"info">( + "Workflow '{}' reloaded and watching", filePath.filename().string()); + else + Logger<"Flow">::template print<"info">( + "Watching new workflow '{}'", filePath.filename().string()); + } + catch (const std::exception& exception) + { + Logger<"Flow">::template print<"warning">( + "Failed to load or reload workflow from '{}': {}", filePath.string(), exception.what()); + } + }); + m_flowsWatcher->watch(m_flowsDirectory); + } + + auto stopWatching() -> void + { + m_flowsWatcher.reset(); + for (const auto& entry : m_flows) + entry.value->stopWatching(); + } + + private: + Thread::ThreadPool& m_threadPool; + std::filesystem::path m_flowsDirectory; + FlowContainer m_flows; + std::unordered_map m_factories; + std::unique_ptr m_flowsWatcher; + }; +} diff --git a/tests/Language/XML/FlowRunner.mpp b/tests/Language/XML/FlowRunner.mpp new file mode 100644 index 00000000..f27e4c09 --- /dev/null +++ b/tests/Language/XML/FlowRunner.mpp @@ -0,0 +1,127 @@ +export module CppUtils.UnitTests.Language.Xml.FlowRunner; + +import std; +import CppUtils; + +namespace CppUtils::UnitTest::Language::Xml::FlowRunner +{ + using namespace std::chrono_literals; + using namespace CppUtils::String::Literals; + using namespace CppUtils::Language::CSV::Literals; + + struct Record final + { + std::string id; + std::string category; + int value = 0; + + auto operator==(const Record&) const -> bool = default; + }; + + using RecordStructMapping = CppUtils::Type::Mapping< + CppUtils::Type::Pair<"Id"_token, &Record::id>, + CppUtils::Type::Pair<"Category"_token, &Record::category>, + CppUtils::Type::Pair<"Value"_token, &Record::value>>; + + struct RecordAlphaMapping final + { + using CSVMapping = CppUtils::Type::Mapping< + CppUtils::Type::Pair<"Id"_token, "A"_excelColumn>, + CppUtils::Type::Pair<"Category"_token, "B"_excelColumn>, + CppUtils::Type::Pair<"Value"_token, "C"_excelColumn>>; + + using StructMapping = RecordStructMapping; + using FunctionMapping = CppUtils::Type::Mapping<>; + }; + + struct RecordBetaMapping final + { + using CSVMapping = CppUtils::Type::Mapping< + CppUtils::Type::Pair<"Value"_token, "A"_excelColumn>, + CppUtils::Type::Pair<"Category"_token, "B"_excelColumn>, + CppUtils::Type::Pair<"Id"_token, "C"_excelColumn>>; + + using StructMapping = RecordStructMapping; + using FunctionMapping = CppUtils::Type::Mapping<>; + }; + + auto _ = TestSuite{"Language/Xml/FlowRunner", + {"Language/Xml/Pipeline"}, + [](TestSuite& suite) { + suite.addTest("Discover and execute flows from directory", [&] { + CppUtils::FileSystem::TemporaryDirectory{[&suite](const auto& temporaryDirectory) -> void { + std::filesystem::create_directories(temporaryDirectory / "flows"); + std::filesystem::create_directories(temporaryDirectory / "input"); + std::filesystem::create_directories(temporaryDirectory / "output"); + + const auto flow1Xml = std::format(R"( + + + + + + + + )", + temporaryDirectory.generic_string(), + temporaryDirectory.generic_string()); + + const auto flow2Xml = std::format(R"( + + + + + + + + )", + temporaryDirectory.generic_string(), + temporaryDirectory.generic_string()); + + CppUtils::FileSystem::String::write(temporaryDirectory / "flows/alpha.xml", flow1Xml); + CppUtils::FileSystem::String::write(temporaryDirectory / "flows/beta.xml", flow2Xml); + + CppUtils::FileSystem::String::write(temporaryDirectory / "input/data1.csv", "rec1;catA;5\nrec2;catB;15\n"); + CppUtils::FileSystem::String::write(temporaryDirectory / "input/data2.csv", "5;catA;rec3\n25;catB;rec4\n"); + + auto threadPool = CppUtils::Thread::ThreadPool{2}; + auto runner = CppUtils::Language::Xml::FlowRunner{threadPool, temporaryDirectory / "flows"}; + runner.registerMapping("alpha"); + runner.registerMapping("beta"); + + suite.expectEqual(std::ranges::size(runner.flows()), 0uz); + runner.loadFlows(); + suite.expectEqual(std::ranges::size(runner.flows()), 2uz); + const auto configs = runner.discoverFlowConfigs(); + suite.expectEqual(std::ranges::size(configs), 2uz); + + const auto alphaFlowByName = runner.flow("FlowAlpha"); + const auto betaFlowByName = runner.flow("FlowBeta"); + suite.expect(alphaFlowByName != nullptr); + suite.expect(betaFlowByName != nullptr); + suite.expect(runner.flow("NonExistent") == nullptr); + + const auto alphaFlowByPath = runner.flow(temporaryDirectory / "flows/alpha.xml"); + const auto betaFlowByPath = runner.flow(temporaryDirectory / "flows/beta.xml"); + suite.expectEqual(alphaFlowByPath, alphaFlowByName); + suite.expectEqual(betaFlowByPath, betaFlowByName); + suite.expect(runner.flow(temporaryDirectory / "flows/missing.xml") == nullptr); + + const auto stats = runner.execute(); + + suite.expectEqual(stats.processedFilesCount, 2uz); + suite.expect(stats.elapsedDuration >= 0ns); + suite.expect(std::filesystem::exists(temporaryDirectory / "output/out1.csv")); + suite.expect(std::filesystem::exists(temporaryDirectory / "output/out2.csv")); + + const auto out1Content = CppUtils::FileSystem::String::read(temporaryDirectory / "output/out1.csv"); + suite.expect(out1Content.contains("rec1;catA;5")); + suite.expect(out1Content.contains("rec2;catB;15")); + + const auto out2Content = CppUtils::FileSystem::String::read(temporaryDirectory / "output/out2.csv"); + suite.expect(out2Content.contains("25;catB;rec4")); + suite.expect(not out2Content.contains("rec3")); + }}; + }); + }}; +} diff --git a/tests/Language/XML/Pipeline.mpp b/tests/Language/XML/Pipeline.mpp index d626bb36..c21c39cc 100644 --- a/tests/Language/XML/Pipeline.mpp +++ b/tests/Language/XML/Pipeline.mpp @@ -721,9 +721,9 @@ Bob;bob@example.com;30;80;4.8)"sv); }); suite.addTest(" with custom loggerName", [&] { - auto loggedMessages = std::make_shared>(); - CppUtils::Logger<"CustomPipelineLogger">::subscribe<"debug">([loggedMessages](const std::string& message) { - loggedMessages->push_back(message); + auto loggedMessages = std::vector{}; + const auto subscriptionIdentifier = CppUtils::Logger<"CustomPipelineLogger">::subscribe<"debug">([&loggedMessages](const std::string& message) { + loggedMessages.push_back(message); }); const auto xmlSource = R"( @@ -744,11 +744,12 @@ Bob;bob@example.com;30;80;4.8)"sv); | std::ranges::to>(); CppUtils::Logger<"CustomPipelineLogger">::waitUntilFinished(); + CppUtils::Logger<"CustomPipelineLogger">::unsubscribe(subscriptionIdentifier); suite.expectEqual(std::ranges::size(outputUsers), 1uz); suite.expectEqual(outputUsers[0].name, "Alice"); - suite.expectEqual(std::ranges::size(*loggedMessages), 1uz); - suite.expectEqual((*loggedMessages)[0], "Processing user"); + suite.expectEqual(std::ranges::size(loggedMessages), 1uz); + suite.expectEqual(loggedMessages[0], "Processing user"); }); suite.addTest("Flow with ThreadPool", [&] { @@ -779,21 +780,30 @@ Bob;bob@example.com;30;80;4.8)"sv); suite.expect(outputContent.contains("Alice")); suite.expect(outputContent.contains("110")); - const auto xmlSourceInMemory = std::format(R"( + auto loggedMessages = std::vector{}; + const auto subscriptionIdentifier = CppUtils::Logger<"FlowWithoutOutputTest">::subscribe<"info">([&loggedMessages](const std::string& message) { + loggedMessages.push_back(message); + }); + + const auto xmlSourceWithoutOutput = std::format(R"( + )", inputPath.generic_string()); - auto flowInMemory = CppUtils::Language::Xml::buildFlow(xmlSourceInMemory, threadPool); - const auto results = flowInMemory.execute(); + auto flowWithoutOutput = CppUtils::Language::Xml::buildFlow(xmlSourceWithoutOutput, threadPool); + flowWithoutOutput.execute(); + + CppUtils::Logger<"FlowWithoutOutputTest">::waitUntilFinished(); + CppUtils::Logger<"FlowWithoutOutputTest">::unsubscribe(subscriptionIdentifier); - suite.expectEqual(std::ranges::size(results), 1uz); - suite.expectEqual(results[0].score, 110); + suite.expectEqual(std::ranges::size(loggedMessages), 1uz); + suite.expect(loggedMessages[0].contains("110")); }}; }); diff --git a/tests/UnitTests.mpp b/tests/UnitTests.mpp index 2e651247..e6d0fcdf 100644 --- a/tests/UnitTests.mpp +++ b/tests/UnitTests.mpp @@ -51,6 +51,7 @@ export import CppUtils.UnitTests.Language.Lexer.Grammar.HighLevelGrammar; export import CppUtils.UnitTests.Language.Markdown.MarkdownLexer; export import CppUtils.UnitTests.Language.Xml.Pipeline; export import CppUtils.UnitTests.Language.Xml.XmlLexer; +export import CppUtils.UnitTests.Language.Xml.FlowRunner; export import CppUtils.UnitTests.Language.VirtualMachine; export import CppUtils.UnitTests.Log.FileSink; export import CppUtils.UnitTests.LogRotate;