Wait on specific events being processed

2 files changed, 11 insertions, 0 deletions

diff --git a/lib/output/pq/updateDatabase.cpp b/lib/output/pq/updateDatabase.cpp
index 485c914..330a0af 100644
--- a/lib/output/pq/updateDatabase.cpp
+++ b/lib/output/pq/updateDatabase.cpp
@@ -183,6 +183,12 @@ namespace MyGrate::Output::Pq {
 		}
 	}
 
+	const EventCounter &
+	UpdateDatabase::getProcessedCounts() const
+	{
+		return processed;
+	}
+
 	void
 	UpdateDatabase::tableMap(MariaDB_Event_Ptr e)
 	{
@@ -207,6 +213,7 @@ namespace MyGrate::Output::Pq {
 	void
 	UpdateDatabase::afterEvent(const MariaDB_Event_Ptr & e)
 	{
+		processed.tick(e->event_type);
 		if (!intx) {
 			output::pq::sql::updateSourcePosition::execute(this, e->next_event_pos, source);
 			commitTx();
diff --git a/lib/output/pq/updateDatabase.h b/lib/output/pq/updateDatabase.h
index 63a96b9..85369fd 100644
--- a/lib/output/pq/updateDatabase.h
+++ b/lib/output/pq/updateDatabase.h
@@ -4,6 +4,7 @@
 #include "pqConn.h"
 #include "pqStmt.h"
 #include <cstdint>
+#include <eventCounter.h>
 #include <eventHandlerBase.h>
 #include <eventSourceBase.h>
 #include <row.h>
@@ -62,6 +63,8 @@ namespace MyGrate::Output::Pq {
 		void gtid(MariaDB_Event_Ptr) override;
 		void xid(MariaDB_Event_Ptr) override;
 
+		const EventCounter & getProcessedCounts() const;
+
 		const uint64_t source;
 		const std::string schema;
 		const std::string database;
@@ -85,6 +88,7 @@ namespace MyGrate::Output::Pq {
 		Tables::const_iterator selected;
 		MariaDB_Event_Ptr table_map;
 		bool intx {false};
+		EventCounter processed;
 	};
 }