Support the rotate event to move to the next binlog file

7 files changed, 19 insertions, 1 deletions

diff --git a/lib/eventHandlerBase.cpp b/lib/eventHandlerBase.cpp
index 07d2c6a..25f0007 100644
--- a/lib/eventHandlerBase.cpp
+++ b/lib/eventHandlerBase.cpp
@@ -6,4 +6,5 @@ namespace MyGrate {
 	void EventHandlerBase::insertRow(MariaDB_Event_Ptr) { }
 	void EventHandlerBase::updateRow(MariaDB_Event_Ptr) { }
 	void EventHandlerBase::deleteRow(MariaDB_Event_Ptr) { }
+	void EventHandlerBase::rotate(MariaDB_Event_Ptr) { }
 }
diff --git a/lib/eventHandlerBase.h b/lib/eventHandlerBase.h
index 8f17e69..07b0c5a 100644
--- a/lib/eventHandlerBase.h
+++ b/lib/eventHandlerBase.h
@@ -13,6 +13,7 @@ namespace MyGrate {
 		virtual void insertRow(MariaDB_Event_Ptr);
 		virtual void updateRow(MariaDB_Event_Ptr);
 		virtual void deleteRow(MariaDB_Event_Ptr);
+		virtual void rotate(MariaDB_Event_Ptr);
 	};
 }
 
diff --git a/lib/eventHandlers.h b/lib/eventHandlers.h
index cefd64f..0fb62ef 100644
--- a/lib/eventHandlers.h
+++ b/lib/eventHandlers.h
@@ -18,7 +18,7 @@ namespace MyGrate {
 		eh[START_EVENT_V3] = {"START_EVENT_V3", nullptr};
 		eh[QUERY_EVENT] = {"QUERY_EVENT", nullptr};
 		eh[STOP_EVENT] = {"STOP_EVENT", nullptr};
-		eh[ROTATE_EVENT] = {"ROTATE_EVENT", nullptr};
+		eh[ROTATE_EVENT] = {"ROTATE_EVENT", &MyGrate::EventHandlerBase::rotate};
 		eh[INTVAR_EVENT] = {"INTVAR_EVENT", nullptr};
 		eh[LOAD_EVENT] = {"LOAD_EVENT", nullptr};
 		eh[SLAVE_EVENT] = {"SLAVE_EVENT", nullptr};
diff --git a/lib/output/pq/sql/updateSourceRotate.sql b/lib/output/pq/sql/updateSourceRotate.sql
new file mode 100644
index 0000000..74e22fc
--- /dev/null
+++ b/lib/output/pq/sql/updateSourceRotate.sql
@@ -0,0 +1,4 @@
+UPDATE mygrate.source SET
+	filename = $1,
+	position = $2
+WHERE source_id = $3
diff --git a/lib/output/pq/updateDatabase.cpp b/lib/output/pq/updateDatabase.cpp
index c43d8e4..9db672d 100644
--- a/lib/output/pq/updateDatabase.cpp
+++ b/lib/output/pq/updateDatabase.cpp
@@ -20,6 +20,7 @@
 #include <output/pq/sql/selectSourceSchema.h>
 #include <output/pq/sql/selectTables.h>
 #include <output/pq/sql/updateSourcePosition.h>
+#include <output/pq/sql/updateSourceRotate.h>
 #include <row.h>
 #include <stdexcept>
 #include <streamSupport.h>
@@ -384,4 +385,13 @@ namespace MyGrate::Output::Pq {
 			afterEvent(e);
 		}
 	}
+
+	void
+	UpdateDatabase::rotate(MariaDB_Event_Ptr e)
+	{
+		if ((e->flags & LOG_EVENT_ARTIFICIAL_F) == 0) {
+			output::pq::sql::updateSourceRotate::execute(
+					this, *e->event.rotate.filename, e->event.rotate.position, source);
+		}
+	}
 }
diff --git a/lib/output/pq/updateDatabase.h b/lib/output/pq/updateDatabase.h
index 9547b90..12df97e 100644
--- a/lib/output/pq/updateDatabase.h
+++ b/lib/output/pq/updateDatabase.h
@@ -57,6 +57,7 @@ namespace MyGrate::Output::Pq {
 		void deleteRow(MariaDB_Event_Ptr) override;
 		void insertRow(MariaDB_Event_Ptr) override;
 		void tableMap(MariaDB_Event_Ptr) override;
+		void rotate(MariaDB_Event_Ptr) override;
 
 		const uint64_t source;
 		const std::string schema;
diff --git a/test/test-e2e.cpp b/test/test-e2e.cpp
index e541ec9..6d484af 100644
--- a/test/test-e2e.cpp
+++ b/test/test-e2e.cpp
@@ -45,6 +45,7 @@ BOOST_AUTO_TEST_CASE(e2e)
 			VALUES(?, ?, ?, ?, ?, now(), now())",
 			5);
 	ins->execute({"hashyhash", "testuser", "groupadm", "10.10.0.1", 2433});
+	mym.query("flush logs");
 
 	sleep(1);