diff options
-rw-r--r-- | lib/Jamfile.jam | 2 | ||||
-rw-r--r-- | lib/eventHandlerBase.cpp | 13 | ||||
-rw-r--r-- | lib/eventHandlerBase.h | 24 | ||||
-rw-r--r-- | lib/eventHandlers.h | 73 | ||||
-rw-r--r-- | lib/eventSourceBase.h | 13 | ||||
-rw-r--r-- | lib/input/mysqlConn.cpp | 19 | ||||
-rw-r--r-- | lib/input/mysqlConn.h | 14 | ||||
-rw-r--r-- | lib/input/replStream.cpp | 32 | ||||
-rw-r--r-- | lib/input/replStream.h | 16 | ||||
-rw-r--r-- | lib/mariadb_repl.h | 11 | ||||
-rw-r--r-- | lib/output/dumpToConsole.cpp | 81 | ||||
-rw-r--r-- | lib/output/dumpToConsole.h | 23 | ||||
-rw-r--r-- | main/main.cpp | 207 |
13 files changed, 325 insertions, 203 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 diff --git a/main/main.cpp b/main/main.cpp index b5d5a4c..bdcda75 100644 --- a/main/main.cpp +++ b/main/main.cpp @@ -1,207 +1,10 @@ -#include <cstddef> -#include <mysql.h> - -#include <array> -#include <bitset> -#include <cmath> -#include <compileTimeFormatter.h> -#include <iomanip> -#include <map> -#include <mariadb_rpl.h> -#include <memory> -#include <vector> - -#include <bitset.h> -#include <rawDataReader.h> -#include <row.h> -#include <streamSupport.h> - -using namespace AdHoc; -using MariaDB_Rpl_Ptr = std::unique_ptr<MARIADB_RPL, decltype(&mariadb_rpl_close)>; -using MariaDB_Event_Ptr = std::unique_ptr<MARIADB_RPL_EVENT, decltype(&mariadb_free_rpl_event)>; - -struct EventHandler { - std::string_view name; - void (*func)(MariaDB_Event_Ptr); -}; -using EventHandlers = std::array<EventHandler, ENUM_END_EVENT>; - -using TableId = decltype(st_mariadb_rpl_table_map_event::table_id); -using TableMaps = std::map<TableId, MariaDB_Event_Ptr>; -TableMaps tableMaps; - -struct write { - template<typename T> - void - operator()(const T & v) const - { - scprintf<"\t\t%?\n">(std::cout, v); - } - void - operator()(const uint8_t & v) const - { - scprintf<"\t\t%d\n">(std::cout, v); - } - void - operator()(const int8_t & v) const - { - scprintf<"\t\t%d\n">(std::cout, v); - } -}; -static void -dumpRowData(const st_mariadb_rpl_rows_event & row) -{ - MyGrate::Row r {row, tableMaps.at(row.table_id)->event.table_map}; - std::for_each(r.begin(), r.end(), [](auto && fv) { - std::visit(write {}, fv); - }); -} -static void -dumpRowPairData(const st_mariadb_rpl_rows_event & row) -{ - MyGrate::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); - }); -} - -static void -doTableMap(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); - // for (auto c = 0U; c < tm.column_types.length; c++) { - // AdHoc::scprintf<"\t%#02hx\n">(std::cout, tm.column_types.str[c]); - //} - tableMaps.insert_or_assign(tm.table_id, std::move(event)); -} - -static void -doWrite(MariaDB_Event_Ptr event) -{ - const auto & rs = event->event.rows; - AdHoc::scprintf<"Insert into %?\n">(std::cout, rs.table_id); - dumpRowData(event->event.rows); -} - -static void -doUpdate(MariaDB_Event_Ptr event) -{ - const auto & rs = event->event.rows; - AdHoc::scprintf<"Update %?\n">(std::cout, rs.table_id); - dumpRowPairData(event->event.rows); -} - -static void -doDelete(MariaDB_Event_Ptr event) -{ - const auto & rs = event->event.rows; - AdHoc::scprintf<"Delete from %?\n">(std::cout, rs.table_id); - dumpRowData(event->event.rows); -} - -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", doTableMap}; - 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", doWrite}; - eh[UPDATE_ROWS_EVENT_V1] = {"UPDATE_ROWS_EVENT_V1", doUpdate}; - eh[DELETE_ROWS_EVENT_V1] = {"DELETE_ROWS_EVENT_V1", doDelete}; - 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; -}()}; - -static void -read_events(MYSQL * mysql) -{ - auto rpl = MariaDB_Rpl_Ptr {mariadb_rpl_init(mysql), &mariadb_rpl_close}; - - mysql_query(mysql, "SET @mariadb_slave_capability=4"); - // mysql_query(mysql, "SET @slave_gtid_strict_mode=1"); - // mysql_query(mysql, "SET @slave_gtid_ignore_duplicates=1"); - mysql_query(mysql, "SET NAMES utf8"); - mysql_query(mysql, "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 (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) { - h.func(std::move(event)); - } - else { - std::cout << h.name << "\n"; - } - } -} +#include <input/replStream.h> +#include <output/dumpToConsole.h> int main(int, char **) { - std::cout << std::boolalpha; - MYSQL conn; - mysql_init(&conn); - if (!mysql_real_connect(&conn, "192.168.1.38", "repl", "r3pl", "", - // NOLINTNEXTLINE(hicpp-signed-bitwise) - 3306, nullptr, CLIENT_LOCAL_FILES | CLIENT_MULTI_STATEMENTS)) { - throw std::runtime_error("ConnectionError"); - } - read_events(&conn); - mysql_close(&conn); + MyGrate::Input::ReplicationStream rs {"192.168.1.38", "repl", "r3pl", 3306}; + MyGrate::Output::DumpToConsole dtc; + rs.readEvents(dtc); } |