summaryrefslogtreecommitdiff
path: root/lib
diff options
context:
space:
mode:
Diffstat (limited to 'lib')
-rw-r--r--lib/Jamfile.jam2
-rw-r--r--lib/eventHandlerBase.cpp13
-rw-r--r--lib/eventHandlerBase.h24
-rw-r--r--lib/eventHandlers.h73
-rw-r--r--lib/eventSourceBase.h13
-rw-r--r--lib/input/mysqlConn.cpp19
-rw-r--r--lib/input/mysqlConn.h14
-rw-r--r--lib/input/replStream.cpp32
-rw-r--r--lib/input/replStream.h16
-rw-r--r--lib/mariadb_repl.h11
-rw-r--r--lib/output/dumpToConsole.cpp81
-rw-r--r--lib/output/dumpToConsole.h23
12 files changed, 320 insertions, 1 deletions
diff --git a/lib/Jamfile.jam b/lib/Jamfile.jam
index 6a31bec..02374c9 100644
--- a/lib/Jamfile.jam
+++ b/lib/Jamfile.jam
@@ -1,5 +1,5 @@
lib mygrate :
- [ glob *.cpp ]
+ [ glob-tree *.cpp ]
:
<link>static
<use>..//libmariadb
diff --git a/lib/eventHandlerBase.cpp b/lib/eventHandlerBase.cpp
new file mode 100644
index 0000000..7f9b2d4
--- /dev/null
+++ b/lib/eventHandlerBase.cpp
@@ -0,0 +1,13 @@
+#include "eventHandlerBase.h"
+
+namespace MyGrate {
+ void
+ EventHandlerBase::tableMap(MyGrate::MariaDB_Event_Ptr event)
+ {
+ tableMaps.insert_or_assign(event->event.table_map.table_id, std::move(event));
+ }
+
+ void EventHandlerBase::insertRow(MariaDB_Event_Ptr) { }
+ void EventHandlerBase::updateRow(MariaDB_Event_Ptr) { }
+ void EventHandlerBase::deleteRow(MariaDB_Event_Ptr) { }
+}
diff --git a/lib/eventHandlerBase.h b/lib/eventHandlerBase.h
new file mode 100644
index 0000000..4bb2396
--- /dev/null
+++ b/lib/eventHandlerBase.h
@@ -0,0 +1,24 @@
+#ifndef MYGRATE_EVENTHANDLERBASE_H
+#define MYGRATE_EVENTHANDLERBASE_H
+
+#include "mariadb_repl.h"
+#include <map>
+#include <memory>
+
+namespace MyGrate {
+ using MariaDB_Event_Ptr = std::unique_ptr<MARIADB_RPL_EVENT, decltype(&mariadb_free_rpl_event)>;
+ class EventHandlerBase {
+ public:
+ virtual void tableMap(MariaDB_Event_Ptr);
+ virtual void insertRow(MariaDB_Event_Ptr);
+ virtual void updateRow(MariaDB_Event_Ptr);
+ virtual void deleteRow(MariaDB_Event_Ptr);
+
+ protected:
+ using TableId = decltype(st_mariadb_rpl_table_map_event::table_id);
+ using TableMaps = std::map<TableId, MyGrate::MariaDB_Event_Ptr>;
+ TableMaps tableMaps;
+ };
+}
+
+#endif
diff --git a/lib/eventHandlers.h b/lib/eventHandlers.h
new file mode 100644
index 0000000..cefd64f
--- /dev/null
+++ b/lib/eventHandlers.h
@@ -0,0 +1,73 @@
+#ifndef MYGRATE_EVENTHANDLERS_H
+#define MYGRATE_EVENTHANDLERS_H
+
+#include "eventHandlerBase.h"
+#include <array>
+#include <string_view>
+
+namespace MyGrate {
+ struct EventHandler {
+ std::string_view name;
+ void (MyGrate::EventHandlerBase::*func)(MyGrate::MariaDB_Event_Ptr);
+ };
+ using EventHandlers = std::array<EventHandler, ENUM_END_EVENT>;
+
+ constexpr EventHandlers eventHandlers {[]() {
+ EventHandlers eh {};
+ eh[UNKNOWN_EVENT] = {"UNKNOWN_EVENT", nullptr};
+ 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[INTVAR_EVENT] = {"INTVAR_EVENT", nullptr};
+ eh[LOAD_EVENT] = {"LOAD_EVENT", nullptr};
+ eh[SLAVE_EVENT] = {"SLAVE_EVENT", nullptr};
+ eh[CREATE_FILE_EVENT] = {"CREATE_FILE_EVENT", nullptr};
+ eh[APPEND_BLOCK_EVENT] = {"APPEND_BLOCK_EVENT", nullptr};
+ eh[EXEC_LOAD_EVENT] = {"EXEC_LOAD_EVENT", nullptr};
+ eh[DELETE_FILE_EVENT] = {"DELETE_FILE_EVENT", nullptr};
+ eh[NEW_LOAD_EVENT] = {"NEW_LOAD_EVENT", nullptr};
+ eh[RAND_EVENT] = {"RAND_EVENT", nullptr};
+ eh[USER_VAR_EVENT] = {"USER_VAR_EVENT", nullptr};
+ eh[FORMAT_DESCRIPTION_EVENT] = {"FORMAT_DESCRIPTION_EVENT", nullptr};
+ eh[XID_EVENT] = {"XID_EVENT", nullptr};
+ eh[BEGIN_LOAD_QUERY_EVENT] = {"BEGIN_LOAD_QUERY_EVENT", nullptr};
+ eh[EXECUTE_LOAD_QUERY_EVENT] = {"EXECUTE_LOAD_QUERY_EVENT", nullptr};
+ eh[TABLE_MAP_EVENT] = {"TABLE_MAP_EVENT", &MyGrate::EventHandlerBase::tableMap};
+ eh[PRE_GA_WRITE_ROWS_EVENT] = {"PRE_GA_WRITE_ROWS_EVENT", nullptr};
+ eh[PRE_GA_UPDATE_ROWS_EVENT] = {"PRE_GA_UPDATE_ROWS_EVENT", nullptr};
+ eh[PRE_GA_DELETE_ROWS_EVENT] = {"PRE_GA_DELETE_ROWS_EVENT", nullptr};
+ eh[WRITE_ROWS_EVENT_V1] = {"WRITE_ROWS_EVENT_V1", &MyGrate::EventHandlerBase::insertRow};
+ eh[UPDATE_ROWS_EVENT_V1] = {"UPDATE_ROWS_EVENT_V1", &MyGrate::EventHandlerBase::updateRow};
+ eh[DELETE_ROWS_EVENT_V1] = {"DELETE_ROWS_EVENT_V1", &MyGrate::EventHandlerBase::deleteRow};
+ eh[INCIDENT_EVENT] = {"INCIDENT_EVENT", nullptr};
+ eh[HEARTBEAT_LOG_EVENT] = {"HEARTBEAT_LOG_EVENT", nullptr};
+ eh[IGNORABLE_LOG_EVENT] = {"IGNORABLE_LOG_EVENT", nullptr};
+ eh[ROWS_QUERY_LOG_EVENT] = {"ROWS_QUERY_LOG_EVENT", nullptr};
+ eh[WRITE_ROWS_EVENT] = {"WRITE_ROWS_EVENT", nullptr};
+ eh[UPDATE_ROWS_EVENT] = {"UPDATE_ROWS_EVENT", nullptr};
+ eh[DELETE_ROWS_EVENT] = {"DELETE_ROWS_EVENT", nullptr};
+ eh[GTID_LOG_EVENT] = {"GTID_LOG_EVENT", nullptr};
+ eh[ANONYMOUS_GTID_LOG_EVENT] = {"ANONYMOUS_GTID_LOG_EVENT", nullptr};
+ eh[PREVIOUS_GTIDS_LOG_EVENT] = {"PREVIOUS_GTIDS_LOG_EVENT", nullptr};
+ eh[TRANSACTION_CONTEXT_EVENT] = {"TRANSACTION_CONTEXT_EVENT", nullptr};
+ eh[VIEW_CHANGE_EVENT] = {"VIEW_CHANGE_EVENT", nullptr};
+ eh[XA_PREPARE_LOG_EVENT] = {"XA_PREPARE_LOG_EVENT", nullptr};
+ eh[MARIA_EVENTS_BEGIN] = {"MARIA_EVENTS_BEGIN", nullptr};
+ eh[ANNOTATE_ROWS_EVENT] = {"ANNOTATE_ROWS_EVENT", nullptr};
+ eh[BINLOG_CHECKPOINT_EVENT] = {"BINLOG_CHECKPOINT_EVENT", nullptr};
+ eh[GTID_EVENT] = {"GTID_EVENT", nullptr};
+ eh[GTID_LIST_EVENT] = {"GTID_LIST_EVENT", nullptr};
+ eh[START_ENCRYPTION_EVENT] = {"START_ENCRYPTION_EVENT", nullptr};
+ eh[QUERY_COMPRESSED_EVENT] = {"QUERY_COMPRESSED_EVENT", nullptr};
+ eh[WRITE_ROWS_COMPRESSED_EVENT_V1] = {"WRITE_ROWS_COMPRESSED_EVENT_V1", nullptr};
+ eh[UPDATE_ROWS_COMPRESSED_EVENT_V1] = {"UPDATE_ROWS_COMPRESSED_EVENT_V1", nullptr};
+ eh[DELETE_ROWS_COMPRESSED_EVENT_V1] = {"DELETE_ROWS_COMPRESSED_EVENT_V1", nullptr};
+ eh[WRITE_ROWS_COMPRESSED_EVENT] = {"WRITE_ROWS_COMPRESSED_EVENT", nullptr};
+ eh[UPDATE_ROWS_COMPRESSED_EVENT] = {"UPDATE_ROWS_COMPRESSED_EVENT", nullptr};
+ eh[DELETE_ROWS_COMPRESSED_EVENT] = {"DELETE_ROWS_COMPRESSED_EVENT", nullptr};
+ return eh;
+ }()};
+}
+
+#endif
diff --git a/lib/eventSourceBase.h b/lib/eventSourceBase.h
new file mode 100644
index 0000000..d511a3f
--- /dev/null
+++ b/lib/eventSourceBase.h
@@ -0,0 +1,13 @@
+#ifndef MYGRATE_EVENTSOURCEBASE_H
+#define MYGRATE_EVENTSOURCEBASE_H
+
+#include "eventHandlerBase.h"
+
+namespace MyGrate {
+ class EventSourceBase {
+ public:
+ virtual void readEvents(EventHandlerBase &) = 0;
+ };
+}
+
+#endif
diff --git a/lib/input/mysqlConn.cpp b/lib/input/mysqlConn.cpp
new file mode 100644
index 0000000..38feb35
--- /dev/null
+++ b/lib/input/mysqlConn.cpp
@@ -0,0 +1,19 @@
+#include "mysqlConn.h"
+#include <stdexcept>
+
+namespace MyGrate::Input {
+ MySQLConn::MySQLConn(const char * const host, const char * const user, const char * const pass, unsigned short port)
+ {
+ mysql_init(this);
+ if (!mysql_real_connect(this, host, user, pass, "", port, nullptr, 0)) {
+ mysql_close(this);
+ throw std::runtime_error("ConnectionError");
+ }
+ mysql_query(this, "SET NAMES utf8");
+ }
+
+ MySQLConn::~MySQLConn()
+ {
+ mysql_close(this);
+ }
+}
diff --git a/lib/input/mysqlConn.h b/lib/input/mysqlConn.h
new file mode 100644
index 0000000..fa3712c
--- /dev/null
+++ b/lib/input/mysqlConn.h
@@ -0,0 +1,14 @@
+#ifndef MYGRATE_INPUT_MYSQLCONN_H
+#define MYGRATE_INPUT_MYSQLCONN_H
+
+#include <mysql.h>
+
+namespace MyGrate::Input {
+ class MySQLConn : public MYSQL {
+ public:
+ MySQLConn(const char * const host, const char * const user, const char * const pass, unsigned short port);
+ ~MySQLConn();
+ };
+}
+
+#endif
diff --git a/lib/input/replStream.cpp b/lib/input/replStream.cpp
new file mode 100644
index 0000000..88c3caf
--- /dev/null
+++ b/lib/input/replStream.cpp
@@ -0,0 +1,32 @@
+#include "replStream.h"
+#include "../eventHandlers.h"
+#include "../mariadb_repl.h"
+
+namespace MyGrate::Input {
+ void
+ ReplicationStream::readEvents(MyGrate::EventHandlerBase & eh)
+ {
+ using MariaDB_Rpl_Ptr = std::unique_ptr<MARIADB_RPL, decltype(&mariadb_rpl_close)>;
+ auto rpl = MariaDB_Rpl_Ptr {mariadb_rpl_init(this), &mariadb_rpl_close};
+
+ mysql_query(this, "SET @mariadb_slave_capability = 4");
+ mysql_query(this, "SET @master_binlog_checksum = @@global.binlog_checksum");
+
+ mariadb_rpl_optionsv(rpl.get(), MARIADB_RPL_SERVER_ID, 12);
+ mariadb_rpl_optionsv(rpl.get(), MARIADB_RPL_FILENAME, "mariadb-bin.000242");
+ mariadb_rpl_optionsv(rpl.get(), MARIADB_RPL_START, 4);
+ mariadb_rpl_optionsv(rpl.get(), MARIADB_RPL_FLAGS, MARIADB_RPL_BINLOG_SEND_ANNOTATE_ROWS);
+
+ if (mariadb_rpl_open(rpl.get())) {
+ throw std::runtime_error("Failed to mariadb_rpl_open");
+ }
+
+ while (MyGrate::MariaDB_Event_Ptr event {mariadb_rpl_fetch(rpl.get(), nullptr), &mariadb_free_rpl_event}) {
+ const auto & h = eventHandlers.at(event->event_type);
+ if (h.func) {
+ (eh.*h.func)(std::move(event));
+ }
+ }
+ }
+
+}
diff --git a/lib/input/replStream.h b/lib/input/replStream.h
new file mode 100644
index 0000000..983d9d3
--- /dev/null
+++ b/lib/input/replStream.h
@@ -0,0 +1,16 @@
+#ifndef MYGRATE_INPUT_REPLSTREAM_H
+#define MYGRATE_INPUT_REPLSTREAM_H
+
+#include "../eventSourceBase.h"
+#include "mysqlConn.h"
+
+namespace MyGrate::Input {
+ class ReplicationStream : public EventSourceBase, MySQLConn {
+ public:
+ using MySQLConn::MySQLConn;
+
+ void readEvents(EventHandlerBase &) override;
+ };
+}
+
+#endif
diff --git a/lib/mariadb_repl.h b/lib/mariadb_repl.h
new file mode 100644
index 0000000..109385e
--- /dev/null
+++ b/lib/mariadb_repl.h
@@ -0,0 +1,11 @@
+#ifndef MYGRATE_MARIADB_REPL_H
+#define MYGRATE_MARIADB_REPL_H
+
+// This file exists because mariadb_rpl.h alone fails as it's missing dependencies
+
+#include <cstddef>
+#include <mysql.h>
+
+#include <mariadb_rpl.h>
+
+#endif
diff --git a/lib/output/dumpToConsole.cpp b/lib/output/dumpToConsole.cpp
new file mode 100644
index 0000000..6ece800
--- /dev/null
+++ b/lib/output/dumpToConsole.cpp
@@ -0,0 +1,81 @@
+#include "dumpToConsole.h"
+#include "../compileTimeFormatter.h"
+#include "../row.h"
+#include "../streamSupport.h"
+
+namespace MyGrate::Output {
+ void
+ DumpToConsole::tableMap(MyGrate::MariaDB_Event_Ptr event)
+ {
+ const auto & tm = event->event.table_map;
+ AdHoc::scprintf<"Table map %?.%? -> %?\n">(std::cout, tm.database, tm.table, tm.table_id);
+ EventHandlerBase::tableMap(std::move(event));
+ }
+
+ void
+ DumpToConsole::insertRow(MyGrate::MariaDB_Event_Ptr event)
+ {
+ const auto & rs = event->event.rows;
+ AdHoc::scprintf<"Insert into %?\n">(std::cout, rs.table_id);
+ dumpRowData(event->event.rows);
+ }
+
+ void
+ DumpToConsole::updateRow(MyGrate::MariaDB_Event_Ptr event)
+ {
+ const auto & rs = event->event.rows;
+ AdHoc::scprintf<"Update %?\n">(std::cout, rs.table_id);
+ dumpRowPairData(event->event.rows);
+ }
+
+ void
+ DumpToConsole::deleteRow(MyGrate::MariaDB_Event_Ptr event)
+ {
+ const auto & rs = event->event.rows;
+ AdHoc::scprintf<"Delete from %?\n">(std::cout, rs.table_id);
+ dumpRowData(event->event.rows);
+ }
+
+ struct write {
+ template<typename T>
+ void
+ operator()(const T & v) const
+ {
+ AdHoc::scprintf<"\t\t%?\n">(std::cout, v);
+ }
+
+ void
+ operator()(const uint8_t & v) const
+ {
+ AdHoc::scprintf<"\t\t%d\n">(std::cout, v);
+ }
+
+ void
+ operator()(const int8_t & v) const
+ {
+ AdHoc::scprintf<"\t\t%d\n">(std::cout, v);
+ }
+ };
+
+ void
+ DumpToConsole::dumpRowData(const st_mariadb_rpl_rows_event & row) const
+ {
+ Row r {row, tableMaps.at(row.table_id)->event.table_map};
+ std::for_each(r.begin(), r.end(), [](auto && fv) {
+ std::visit(write {}, fv);
+ });
+ }
+
+ void
+ DumpToConsole::dumpRowPairData(const st_mariadb_rpl_rows_event & row) const
+ {
+ RowPair rp {row, tableMaps.at(row.table_id)->event.table_map};
+ std::for_each(rp.first.begin(), rp.first.end(), [](auto && fv) {
+ std::visit(write {}, fv);
+ });
+ std::for_each(rp.second.begin(), rp.second.end(), [](auto && fv) {
+ std::visit(write {}, fv);
+ });
+ }
+
+}
diff --git a/lib/output/dumpToConsole.h b/lib/output/dumpToConsole.h
new file mode 100644
index 0000000..98769a6
--- /dev/null
+++ b/lib/output/dumpToConsole.h
@@ -0,0 +1,23 @@
+#ifndef MYGRATE_OUTPUT_DUMPTOCONSOLE_H
+#define MYGRATE_OUTPUT_DUMPTOCONSOLE_H
+
+#include "../eventHandlerBase.h"
+
+namespace MyGrate::Output {
+ class DumpToConsole : public MyGrate::EventHandlerBase {
+ protected:
+ void tableMap(MyGrate::MariaDB_Event_Ptr event) override;
+
+ void insertRow(MyGrate::MariaDB_Event_Ptr event) override;
+
+ void updateRow(MyGrate::MariaDB_Event_Ptr event) override;
+
+ void deleteRow(MyGrate::MariaDB_Event_Ptr event) override;
+
+ private:
+ void dumpRowData(const st_mariadb_rpl_rows_event & row) const;
+ void dumpRowPairData(const st_mariadb_rpl_rows_event & row) const;
+ };
+}
+
+#endif
generated by cgit v1.2.3 (git 2.25.1) at 2024-11-22 07:29:45 +0000