1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
|
#include "replStream.h"
#include "mariadb_repl.h"
#include "mysqlConn.h"
#include <eventHandlerBase.h>
#include <eventHandlers.h>
#include <helpers.h>
#include <memory>
#include <stdexcept>
#include <utility>
namespace MyGrate::Input {
ReplicationStream::ReplicationStream(const std::string & host, const std::string & user, const std::string & pass,
unsigned short port, uint64_t sid, std::string fn, uint64_t pos) :
MySQLConn {host.c_str(), user.c_str(), pass.c_str(), port},
serverid {sid}, filename {std::move(fn)}, position {pos}
{
}
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};
query("SET @mariadb_slave_capability = 4");
query("SET @master_binlog_checksum = @@global.binlog_checksum");
mariadb_rpl_optionsv(rpl.get(), MARIADB_RPL_SERVER_ID, serverid);
mariadb_rpl_optionsv(rpl.get(), MARIADB_RPL_FILENAME, filename.c_str(), filename.length());
mariadb_rpl_optionsv(rpl.get(), MARIADB_RPL_START, position);
mariadb_rpl_optionsv(rpl.get(), MARIADB_RPL_FLAGS, MARIADB_RPL_BINLOG_SEND_ANNOTATE_ROWS);
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}) {
position = event->next_event_pos;
if (const auto & h = eventHandlers.at(event->event_type); h.func) {
(eh.*h.func)(std::move(event));
}
}
}
void
ReplicationStream::stopEvents()
{
mariadb_cancel(this);
}
}
|
