From 172567f6ee168846b41d11ed7b8bd9c1ccbb659b Mon Sep 17 00:00:00 2001
From: Dan Goodliffe <dan@randomdan.homeip.net>
Date: Sun, 22 Aug 2021 13:58:49 +0100
Subject: Wait on specific events being processed
---
lib/input/replStream.cpp | 2 ++
lib/input/replStream.h | 5 +++++
lib/output/pq/updateDatabase.cpp | 7 +++++++
lib/output/pq/updateDatabase.h | 4 ++++
lib/streamSupport.h | 1 +
5 files changed, 19 insertions(+)
(limited to 'lib')
diff --git a/lib/input/replStream.cpp b/lib/input/replStream.cpp
index d04e16b..3bfe877 100644
--- a/lib/input/replStream.cpp
+++ b/lib/input/replStream.cpp
@@ -39,8 +39,10 @@ namespace MyGrate::Input {
verify<MySQLErr>(!mariadb_rpl_open(rpl.get()), "Failed to mariadb_rpl_open", this);
while (MyGrate::MariaDB_Event_Ptr event {mariadb_rpl_fetch(rpl.get(), nullptr), &mariadb_free_rpl_event}) {
+ received.tick(event->event_type);
auto np = event->next_event_pos;
if (const auto & h = eventHandlers.at(event->event_type); h.func) {
+ handled.tick(event->event_type);
(eh.*h.func)(std::move(event));
}
position = np;
diff --git a/lib/input/replStream.h b/lib/input/replStream.h
index 40ec390..4958459 100644
--- a/lib/input/replStream.h
+++ b/lib/input/replStream.h
@@ -3,6 +3,7 @@
#include "mysqlConn.h"
#include <cstdint>
+#include <eventCounter.h>
#include <eventSourceBase.h>
#include <string>
@@ -19,10 +20,14 @@ namespace MyGrate::Input {
void readEvents(EventHandlerBase &) override;
void stopEvents() override;
+ const EventCounter & getReceivedCounts() const;
+ const EventCounter & getHandledCounts() const;
+
private:
uint64_t serverid;
std::string filename;
uint64_t position;
+ EventCounter received, handled;
};
}
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;
};
}
diff --git a/lib/streamSupport.h b/lib/streamSupport.h
index cab62e6..e9af010 100644
--- a/lib/streamSupport.h
+++ b/lib/streamSupport.h
@@ -1,6 +1,7 @@
#ifndef MYGRATE_STREAM_SUPPORT_H
#define MYGRATE_STREAM_SUPPORT_H
+#include "eventCounter.h"
#include "mariadb_repl.h"
#include <array>
#include <cstddef>
--
cgit v1.2.3