Return a callable from jobs

Allows safely running finalisation code in the main thread if required.

3 files changed, 27 insertions, 20 deletions

diff --git a/src/ingestor.cpp b/src/ingestor.cpp
index 73ee239..9159a12 100644
--- a/src/ingestor.cpp
+++ b/src/ingestor.cpp
@@ -368,7 +368,7 @@ namespace WebStat {
 	Ingestor::finalizeJob(Job & job, const std::chrono::minutes freq, const Job::LastRunTime::clock::time_point now)
 	{
 		try {
-			job.currentRun->get();
+			job.currentRun->get()();
 			job.lastRun = now;
 		}
 		catch (const std::exception & excp) {
@@ -409,7 +409,7 @@ namespace WebStat {
 		finishJob(purgeOldLogs);
 	}
 
-	unsigned int
+	Ingestor::Job::Result
 	Ingestor::jobIngestParkedLines()
 	{
 		unsigned int count = 0;
@@ -420,7 +420,9 @@ namespace WebStat {
 				count += 1;
 			}
 		}
-		return count;
+		return [count]() {
+			return count;
+		};
 	}
 
 	void
@@ -458,7 +460,7 @@ namespace WebStat {
 		return count;
 	}
 
-	unsigned int
+	Ingestor::Job::Result
 	Ingestor::jobPurgeOldLogs()
 	{
 		auto dbconn = dbpool->get();
@@ -475,7 +477,9 @@ namespace WebStat {
 			}
 			std::this_thread::sleep_for(settings.purgeDeletePause);
 		}
-		return purgedTotal;
+		return [purgedTotal]() {
+			return purgedTotal;
+		};
 	}
 
 	void
diff --git a/src/ingestor.hpp b/src/ingestor.hpp
index c57c16a..4882b7d 100644
--- a/src/ingestor.hpp
+++ b/src/ingestor.hpp
@@ -60,8 +60,20 @@ namespace WebStat {
 		std::expected<std::filesystem::path, int> parkQueuedLogLines();
 		void runJobsAsNeeded();
 
-		unsigned int jobIngestParkedLines();
-		unsigned int jobPurgeOldLogs();
+		struct Job {
+			using LastRunTime = std::chrono::system_clock::time_point;
+			using Result = std::function<unsigned int()>;
+			using Impl = Result (Ingestor::*)();
+
+			explicit Job(Impl jobImpl) : impl(jobImpl) { }
+
+			const Impl impl;
+			LastRunTime lastRun {LastRunTime::clock::now()};
+			std::optional<std::future<Result>> currentRun;
+		};
+
+		Job::Result jobIngestParkedLines();
+		Job::Result jobPurgeOldLogs();
 
 		template<typename... T> void storeLogLine(DB::Connection *, const std::tuple<T...> &) const;
 
@@ -86,17 +98,6 @@ namespace WebStat {
 
 		bool terminated = false;
 
-		struct Job {
-			using LastRunTime = std::chrono::system_clock::time_point;
-			using Impl = unsigned int (Ingestor::*)();
-
-			explicit Job(Impl jobImpl) : impl(jobImpl) { }
-
-			const Impl impl;
-			LastRunTime lastRun {LastRunTime::clock::now()};
-			std::optional<std::future<unsigned int>> currentRun;
-		};
-
 		Job::LastRunTime lastCheckedJobs {Job::LastRunTime::clock::now()};
 		Job ingestParkedLines;
 		Job purgeOldLogs;
diff --git a/test/test-ingest.cpp b/test/test-ingest.cpp
index 88e80e8..4f9c7ab 100644
--- a/test/test-ingest.cpp
+++ b/test/test-ingest.cpp
@@ -320,7 +320,9 @@ BOOST_AUTO_TEST_CASE(IngestParked, *boost::unit_test::depends_on("I/ParkLogLine"
 	BOOST_REQUIRE(parkQueuedLogLines());
 	BOOST_CHECK(!std::filesystem::is_empty(settings.fallbackDir));
 	BOOST_REQUIRE(queuedLines.empty());
-	jobIngestParkedLines();
+	const auto result = jobIngestParkedLines();
+	BOOST_REQUIRE(result);
+	BOOST_CHECK_EQUAL(result(), 1);
 	BOOST_CHECK_EQUAL(queuedLines.size(), 2);
 	BOOST_CHECK(std::filesystem::is_empty(settings.fallbackDir));
 }
@@ -418,7 +420,7 @@ BOOST_AUTO_TEST_CASE(DiscardUnparsable)
 
 BOOST_AUTO_TEST_CASE(PurgeOldJob)
 {
-	BOOST_CHECK_EQUAL(2, jobPurgeOldLogs());
+	BOOST_CHECK_EQUAL(2, jobPurgeOldLogs()());
 }
 
 BOOST_AUTO_TEST_CASE(LogStatsSignal)