feat(filesystem): make Watcher passive, rearmable, movable and embed as direct member

modules/FileSystem/Watcher.mpp

@@ -41,14 +41,15 @@ export namespace CppUtils::FileSystem
 	public:
 		using SubscribeFunction = std::function<void(Event, const std::filesystem::path&)>;
 
-#if not defined(OS_MACOS)
+#if defined(OS_MACOS)
 		inline Watcher():
-			m_threadLoop { std::bind(&Watcher::listener, this), std::bind(&Watcher::interruptFunction, this) }
+			m_dispatchQueue{dispatch_queue_create("CppUtils.FileSystem.Watcher", nullptr)}
+		{}
 #else
-		inline Watcher()
-#endif
+		inline Watcher():
+			m_threadLoop{[this] { listener(); }, [this] { interruptFunction(); }}
 		{
-#if defined(OS_LINUX)
+#	if defined(OS_LINUX)
 			m_eventPoll = epoll_create1(0);
 			{
 				m_inotify = inotify_init1(IN_NONBLOCK);
@@ -65,13 +66,90 @@ export namespace CppUtils::FileSystem
 					epoll_ctl(m_eventPoll, EPOLL_CTL_ADD, m_stopPipe.read, std::addressof(m_stopEvent)),
 					"Can't add pipe file descriptor to epoll");
 			}
-#elif defined(OS_WINDOWS)
+#	elif defined(OS_WINDOWS)
 			m_stopEvent = CreateEventW(nullptr, true, false, nullptr);
 			m_wakeEvent = CreateEventW(nullptr, true, false, nullptr);
+#	endif
+		}
+#endif
+
+		Watcher(const Watcher&) = delete;
+		auto operator=(const Watcher&) -> Watcher& = delete;
+
+#if defined(OS_MACOS)
+		inline Watcher(Watcher&& other) noexcept
+		{
+			const auto wasRunning = other.isRunning();
+			other.stop();
+			m_dispatchQueue = std::exchange(other.m_dispatchQueue, nullptr);
+			m_isRunning.store(other.m_isRunning.load());
+			m_subscribedFunctions = std::move(other.m_subscribedFunctions);
+			m_watchedItems = std::move(other.m_watchedItems);
+
+			if (wasRunning)
+				start();
+		}
+#else
+		inline Watcher(Watcher&& other) noexcept:
+			m_threadLoop{[this] { listener(); }, [this] { interruptFunction(); }}
+		{
+			const auto wasRunning = other.isRunning();
+			other.stop();
+#	if defined(OS_LINUX)
+			m_eventPoll = std::move(other.m_eventPoll);
+			m_inotify = std::move(other.m_inotify);
+			m_stopPipe = std::move(other.m_stopPipe);
+			m_inotifyEpollEvent = other.m_inotifyEpollEvent;
+			m_stopEvent = other.m_stopEvent;
+			m_events = other.m_events;
+#	elif defined(OS_WINDOWS)
+			m_stopEvent = std::exchange(other.m_stopEvent, nullptr);
+			m_wakeEvent = std::exchange(other.m_wakeEvent, nullptr);
+#	endif
+			m_subscribedFunctions = std::move(other.m_subscribedFunctions);
+			m_watchedItems = std::move(other.m_watchedItems);
+
+			if (wasRunning)
+				start();
+		}
+#endif
+
+		inline auto operator=(Watcher&& other) noexcept -> Watcher&
+		{
+			if (this == std::addressof(other))
+				return *this;
+
+			stop();
+			const auto wasRunning = other.isRunning();
+			other.stop();
+
+#if defined(OS_LINUX)
+			m_eventPoll = std::move(other.m_eventPoll);
+			m_inotify = std::move(other.m_inotify);
+			m_stopPipe = std::move(other.m_stopPipe);
+			m_inotifyEpollEvent = other.m_inotifyEpollEvent;
+			m_stopEvent = other.m_stopEvent;
+			m_events = other.m_events;
+#elif defined(OS_WINDOWS)
+			if (m_stopEvent != nullptr)
+				CloseHandle(m_stopEvent);
+			if (m_wakeEvent != nullptr)
+				CloseHandle(m_wakeEvent);
+			m_stopEvent = std::exchange(other.m_stopEvent, nullptr);
+			m_wakeEvent = std::exchange(other.m_wakeEvent, nullptr);
 #elif defined(OS_MACOS)
-			m_dispatchQueue = dispatch_queue_create("CppUtils.FileSystem.Watcher", nullptr);
+			if (m_dispatchQueue != nullptr)
+				dispatch_release(m_dispatchQueue);
+			m_dispatchQueue = std::exchange(other.m_dispatchQueue, nullptr);
+			m_isRunning.store(other.m_isRunning.load());
 #endif
-			start();
+			m_subscribedFunctions = std::move(other.m_subscribedFunctions);
+			m_watchedItems = std::move(other.m_watchedItems);
+
+			if (wasRunning)
+				start();
+
+			return *this;
 		}
 
 		~Watcher()
@@ -105,10 +183,15 @@ export namespace CppUtils::FileSystem
 
 		inline auto start() -> void
 		{
-#if defined(OS_MACOS)
-			m_isRunning.store(true);
-#else
+#if defined(OS_LINUX)
+			drainStopPipe();
+			m_threadLoop.start();
+#elif defined(OS_WINDOWS)
+			if (m_stopEvent != nullptr)
+				ResetEvent(m_stopEvent);
 			m_threadLoop.start();
+#elif defined(OS_MACOS)
+			m_isRunning.store(true);
 #endif
 		}
 
@@ -118,9 +201,22 @@ export namespace CppUtils::FileSystem
 			m_isRunning.store(false);
 #else
 			m_threadLoop.stop();
+#	if defined(OS_LINUX)
+			drainStopPipe();
+#	elif defined(OS_WINDOWS)
+			if (m_stopEvent != nullptr)
+				ResetEvent(m_stopEvent);
+#	endif
 #endif
 		}
 
+		inline auto reset() -> void
+		{
+			stop();
+			unwatchAll();
+			m_subscribedFunctions.access()->clear();
+		}
+
 #if not defined(OS_MACOS)
 		inline auto listener() -> void
 		{
@@ -232,7 +328,10 @@ export namespace CppUtils::FileSystem
 				{
 					auto eventFileDescriptor = m_events[i].data.fd;
 					if (eventFileDescriptor == m_stopPipe.read)
+					{
+						drainStopPipe();
 						continue;
+					}
 					length = read(eventFileDescriptor, std::ranges::data(buffer), std::ranges::size(buffer));
 					if (length == -1 and errno == EINTR)
 						break;
@@ -276,6 +375,9 @@ export namespace CppUtils::FileSystem
 			if (not std::filesystem::exists(path))
 				return;
 
+			if (not isRunning())
+				start();
+
 			const auto normalizedPath = std::filesystem::weakly_canonical(path);
 
 #if defined(OS_LINUX)
@@ -432,9 +534,8 @@ export namespace CppUtils::FileSystem
 			auto itemsAccessor = m_watchedItems.access();
 #if defined(OS_LINUX)
 			for (const auto& [path, watchDescriptor] : itemsAccessor.value())
-				System::throwErrno(
-					inotify_rm_watch(m_inotify, std::to_underlying(watchDescriptor)),
-					std::format("Failed to stop watching the file {}", path.string()));
+				if (inotify_rm_watch(m_inotify, std::to_underlying(watchDescriptor)) == -1 and errno != EINVAL)
+					System::throwErrno(-1, std::format("Failed to stop watching the file {}", path.string()));
 #elif defined(OS_WINDOWS)
 			for (auto& item : itemsAccessor.value())
 			{
@@ -477,6 +578,15 @@ export namespace CppUtils::FileSystem
 				SetEvent(m_stopEvent);
 #	endif
 		}
+
+#	if defined(OS_LINUX)
+		inline auto drainStopPipe() -> void
+		{
+			auto buffer = std::array<std::byte, 64>{};
+			while (read(m_stopPipe.read, std::ranges::data(buffer), std::ranges::size(buffer)) > 0)
+			{}
+		}
+#	endif
 #endif
 
 #if defined(OS_MACOS)

modules/Language/XML/Flow.mpp

@@ -121,26 +121,22 @@ export namespace CppUtils::Language::Xml
 
 		auto startWatching() -> void override
 		{
-			if (not m_sourceDirectory.has_value() or m_watcher)
+			if (not m_sourceDirectory.has_value() or m_watcher.isRunning())
 				return;
 
 			if (m_outputDirectory.has_value())
 				std::filesystem::create_directories(m_outputDirectory.value());
 
-			m_watcher = std::make_unique<FileSystem::Watcher>();
-			m_watcher->onEvent([this](auto, const std::filesystem::path& filePath) {
+			m_watcher.reset();
+			m_watcher.onEvent([this](auto, const std::filesystem::path& filePath) {
 				if (matchesPattern(filePath))
 					processFileWithStaging(filePath, computeOutputFile(filePath));
 			});
-			m_watcher->watch(m_sourceDirectory.value());
+			m_watcher.watch(m_sourceDirectory.value());
 		}
 
 		auto stopWatching() -> void override
 		{
-			if (not m_watcher)
-				return;
-
-			m_watcher->stop();
 			m_watcher.reset();
 		}
 
@@ -294,7 +290,7 @@ export namespace CppUtils::Language::Xml
 		std::reference_wrapper<Thread::ThreadPool> m_threadPool;
 		bool m_watch = false;
 
-		std::unique_ptr<FileSystem::Watcher> m_watcher;
+		FileSystem::Watcher m_watcher;
 		std::unordered_map<Type::Token, TagHandler> m_tagRegistry;
 		std::unordered_map<Type::Token, FormatProcessor> m_formatRegistry;
 		std::vector<Container::Tree::VariantNode<Type::Token, std::string>> m_pipelineNodes;

modules/Language/XML/FlowRunner.mpp

@@ -195,8 +195,8 @@ export namespace CppUtils::Language::Xml
 			for (const auto& entry : m_flows)
 				entry.value->startWatching();
 
-			m_flowsWatcher = std::make_unique<FileSystem::Watcher>();
-			m_flowsWatcher->onEvent([this](FileSystem::Event event, const std::filesystem::path& filePath) {
+			m_flowsWatcher.reset();
+			m_flowsWatcher.onEvent([this](FileSystem::Event event, const std::filesystem::path& filePath) {
 				if (filePath.extension() != ".xml")
 					return;
 
@@ -246,7 +246,7 @@ export namespace CppUtils::Language::Xml
 							"Failed to load or reload workflow from '{}': {}", filePath.string(), exception.what());
 					}
 			});
-			m_flowsWatcher->watch(m_flowsDirectory);
+			m_flowsWatcher.watch(m_flowsDirectory);
 		}
 
 		auto stopWatching() -> void
@@ -261,6 +261,6 @@ export namespace CppUtils::Language::Xml
 		std::filesystem::path m_flowsDirectory;
 		FlowContainer m_flows;
 		std::unordered_map<std::string, FlowFactory> m_factories;
-		std::unique_ptr<FileSystem::Watcher> m_flowsWatcher;
+		FileSystem::Watcher m_flowsWatcher;
 	};
 }

tests/FileSystem/Watcher.mpp

@@ -174,5 +174,52 @@ namespace CppUtils::UnitTest::FileSystem::Watcher
 				suite.expectWait([&] { return expectDeleteFile.load(); });
 			}};
 		});
+
+		suite.addTest("Passive start and stop", [&] {
+			auto watcher = CppUtils::FileSystem::Watcher{};
+			suite.expect(not watcher.isRunning());
+
+			watcher.start();
+			suite.expect(watcher.isRunning());
+
+			watcher.stop();
+			suite.expect(not watcher.isRunning());
+
+			watcher.start();
+			suite.expect(watcher.isRunning());
+
+			watcher.stop();
+			suite.expect(not watcher.isRunning());
+		});
+
+		suite.addTest("Reset watcher", [&] {
+			CppUtils::FileSystem::TemporaryDirectory{[&suite](const auto& directory) -> void {
+				auto watcher = CppUtils::FileSystem::Watcher{};
+				suite.expect(not watcher.isRunning());
+
+				watcher.watch(directory);
+				suite.expect(watcher.isRunning());
+
+				watcher.reset();
+				suite.expect(not watcher.isRunning());
+			}};
+		});
+
+		suite.addTest("Move watcher", [&] {
+			CppUtils::FileSystem::TemporaryDirectory{[&suite](const auto& directory) -> void {
+				auto watcher1 = CppUtils::FileSystem::Watcher{};
+				suite.expect(not watcher1.isRunning());
+
+				watcher1.watch(directory);
+				suite.expect(watcher1.isRunning());
+
+				auto watcher2 = std::move(watcher1);
+				suite.expect(watcher2.isRunning());
+				suite.expect(not watcher1.isRunning());
+
+				watcher2.stop();
+				suite.expect(not watcher2.isRunning());
+			}};
+		});
 	}};
 }