Wait on specific events being processed

3 files changed, 26 insertions, 75 deletions

diff --git a/test/semaphore.cpp b/test/semaphore.cpp
deleted file mode 100644
index b7e089f..0000000
--- a/test/semaphore.cpp
+++ /dev/null
@@ -1,37 +0,0 @@
-#include "semaphore.h"
-
-#ifndef __cpp_lib_semaphore
-#	include <thread>
-
-semaphore::semaphore(unsigned int v_) : v {v_} { }
-
-void
-semaphore::release(unsigned int n)
-{
-	std::lock_guard lk {m};
-	v += n;
-}
-
-void
-semaphore::acquire()
-{
-	while (!try_dec()) { }
-}
-
-bool
-semaphore::try_dec()
-{
-	m.lock();
-	if (v) {
-		v--;
-		m.unlock();
-		return true;
-	}
-	else {
-		m.unlock();
-		std::this_thread::sleep_for(std::chrono::milliseconds(100));
-		return false;
-	}
-}
-
-#endif
diff --git a/test/semaphore.h b/test/semaphore.h
deleted file mode 100644
index 6a5963f..0000000
--- a/test/semaphore.h
+++ /dev/null
@@ -1,27 +0,0 @@
-#ifndef MYGRATE_TEST_SEMAPHORE_H
-#define MYGRATE_TEST_SEMAPHORE_H
-
-#if __has_include(<semaphore>)
-#	include <semaphore>
-#endif
-
-#ifdef __cpp_lib_semaphore
-using semaphore = std::binary_semaphore;
-#else
-#	include <mutex>
-class semaphore {
-public:
-	explicit semaphore(unsigned int v_);
-
-	void release(unsigned int n = 1);
-
-	void acquire();
-
-private:
-	bool try_dec();
-	unsigned int v;
-	std::mutex m;
-};
-#endif
-
-#endif
diff --git a/test/test-e2e.cpp b/test/test-e2e.cpp
index 4dd40d7..45a8d49 100644
--- a/test/test-e2e.cpp
+++ b/test/test-e2e.cpp
@@ -2,10 +2,11 @@
 #include <boost/test/data/test_case.hpp>
 #include <boost/test/unit_test.hpp>
 
-#include "semaphore.h"
+#include "helpers.h"
 #include "testdb-mysql.h"
 #include "testdb-postgresql.h"
 #include <compileTimeFormatter.h>
+#include <condition_variable>
 #include <dbConn.h>
 #include <input/replStream.h>
 #include <output/pq/updateDatabase.h>
@@ -26,20 +27,25 @@ public:
 	void
 	afterEvent(const MyGrate::MariaDB_Event_Ptr & e) override
 	{
-		UpdateDatabase::afterEvent(std::move(e));
-		ops.release();
+		{
+			std::lock_guard<std::mutex> lk(m);
+			UpdateDatabase::afterEvent(std::move(e));
+		}
+		cv.notify_all();
 	}
 
 	void
-	waitFor(unsigned int n)
+	waitFor(EventCounterTarget ect)
 	{
-		while (n--) {
-			ops.acquire();
-		}
+		std::unique_lock<std::mutex> lk(m);
+		cv.wait(lk, [&] {
+			return getProcessedCounts() >= ect;
+		});
 	}
 
 private:
-	semaphore ops {0};
+	std::condition_variable cv;
+	std::mutex m;
 };
 
 using namespace MyGrate::Testing;
@@ -80,7 +86,7 @@ public:
 	}
 
 	void
-	stopAfter(unsigned int events)
+	stopAfter(const EventCounterTarget & events)
 	{
 		BOOST_REQUIRE(out);
 		BOOST_REQUIRE(src);
@@ -126,7 +132,11 @@ BOOST_AUTO_TEST_CASE(e2e, *boost::unit_test::timeout(15))
 	ins->execute({"hashyhash", "testuser", "groupadm", "10.10.0.1", 2433});
 	mym.query("flush logs");
 
-	stopAfter(4);
+	stopAfter(EventCounterTarget {}
+					  .add(UPDATE_ROWS_EVENT_V1, 1)
+					  .add(DELETE_ROWS_EVENT_V1, 1)
+					  .add(WRITE_ROWS_EVENT_V1, 1)
+					  .add(ROTATE_EVENT, 1));
 }
 
 BOOST_AUTO_TEST_CASE(txns, *boost::unit_test::timeout(15))
@@ -145,7 +155,12 @@ BOOST_AUTO_TEST_CASE(txns, *boost::unit_test::timeout(15))
 		MyGrate::sql::simpleUpdateAll::execute(&mym, "Same");
 		MyGrate::sql::simpleDeleteSome::execute(&mym, 5);
 	});
-	stopAfter(14);
+	stopAfter(EventCounterTarget {}
+					  .add(GTID_EVENT, 1)
+					  .add(WRITE_ROWS_EVENT_V1, 10)
+					  .add(UPDATE_ROWS_EVENT_V1, 1)
+					  .add(DELETE_ROWS_EVENT_V1, 1)
+					  .add(XID_EVENT, 1));
 
 	auto recs = MyGrate::sql::simpleSelectAll::execute(&pqm);
 	BOOST_REQUIRE_EQUAL(recs->rows(), 6);