Skip to content

Commit 33c7e09

Browse files
author
pryanikov
committed
another refuktoring
1 parent 8060091 commit 33c7e09

9 files changed

+363
-360
lines changed

‎dbreader.cpp

+76-46
Original file line numberDiff line numberDiff line change
@@ -6,23 +6,9 @@ using std::placeholders::_1;
66

77
namespace replicator {
88

9-
DBReader::DBReader(nanomysql::mysql_conn_opts &opts, unsigned connect_retry)
10-
: state(), slave(slave::MasterInfo(opts, connect_retry), state), stopped(false), last_event_when(0) {}
11-
12-
DBReader::~DBReader()
13-
{
14-
slave.close_connection();
15-
}
16-
17-
void DBReader::AddTable(const std::string& db, const std::string& table, std::map<std::string, unsigned>& filter, bool do_dump)
18-
{
19-
tables.emplace_back(db, table, filter, do_dump);
20-
}
21-
22-
void DBReader::DumpTables(std::string &binlog_name, unsigned long &binlog_pos, BinlogEventCallback cb)
9+
void DBReader::DumpTables(std::string& binlog_name, unsigned long& binlog_pos, const BinlogEventCallback& cb)
2310
{
24-
slave::callback dummycallback = std::bind(&DBReader::DummyEventCallback, this, _1);
25-
11+
const static slave::callback dummycallback = [] (const slave::RecordSet& event) {};
2612
// start temp slave to read DB structure
2713
slave::Slave tempslave(slave.masterInfo(), state);
2814
for (auto i = tables.begin(), end = tables.end(); i != end; ++i) {
@@ -32,7 +18,7 @@ void DBReader::DumpTables(std::string &binlog_name, unsigned long &binlog_pos, B
3218
tempslave.init();
3319
tempslave.createDatabaseStructure();
3420

35-
last_event_when = ::time(NULL);
21+
// last_event_when = ::time(NULL);
3622

3723
slave::Slave::binlog_pos_t bp = tempslave.getLastBinlog();
3824
binlog_name = bp.first;
@@ -42,12 +28,9 @@ void DBReader::DumpTables(std::string &binlog_name, unsigned long &binlog_pos, B
4228

4329
// dump tables
4430
nanomysql::Connection conn(slave.masterInfo().conn_options);
45-
4631
conn.query("SET NAMES utf8");
4732

4833
for (auto table = tables.begin(), end = tables.end(); !stopped && table != end; ++table) {
49-
if (!table->do_dump) continue;
50-
5134
// build field_name -> field_ptr map for filtered columns
5235
std::vector<std::pair<unsigned, slave::PtrField>> filter_;
5336
const auto rtable = tempslave.getRli().getTable(table->name);
@@ -56,7 +39,7 @@ void DBReader::DumpTables(std::string &binlog_name, unsigned long &binlog_pos, B
5639
for (const auto ptr_field : rtable->fields) {
5740
const auto it = table->filter.find(ptr_field->field_name);
5841
if (it != table->filter.end()) {
59-
filter_.emplace_back(it->second, ptr_field);
42+
filter_.emplace_back(it->second.first, ptr_field);
6043
s_fields += ptr_field->field_name;
6144
s_fields += ',';
6245
}
@@ -71,7 +54,7 @@ void DBReader::DumpTables(std::string &binlog_name, unsigned long &binlog_pos, B
7154
std::cref(table->name.second),
7255
std::cref(filter_),
7356
_1,
74-
cb
57+
std::cref(cb)
7558
));
7659
}
7760

@@ -81,15 +64,15 @@ void DBReader::DumpTables(std::string &binlog_name, unsigned long &binlog_pos, B
8164
ev->binlog_name = binlog_name;
8265
ev->binlog_pos = binlog_pos;
8366
// ev->seconds_behind_master = GetSecondsBehindMaster();
84-
ev->unix_timestamp = long(time(NULL));
67+
// ev->unix_timestamp = long(time(NULL));
8568
ev->event = "IGNORE";
8669
stopped = cb(std::move(ev));
8770
}
8871

8972
//tempslave.close_connection();
9073
}
9174

92-
void DBReader::ReadBinlog(const std::string &binlog_name, unsigned long binlog_pos, BinlogEventCallback cb)
75+
void DBReader::ReadBinlog(const std::string& binlog_name, unsigned long binlog_pos, const BinlogEventCallback& cb)
9376
{
9477
stopped = false;
9578
state.setMasterLogNamePos(binlog_name, binlog_pos);
@@ -98,10 +81,15 @@ void DBReader::ReadBinlog(const std::string &binlog_name, unsigned long binlog_p
9881
slave.setCallback(
9982
table->name.first,
10083
table->name.second,
101-
std::bind(&DBReader::EventCallback, this, _1, std::cref(table->filter), cb)
84+
std::bind(
85+
table->is_primary
86+
? &DBReader::EventCallbackNormal
87+
: &DBReader::EventCallbackNullify,
88+
this, _1, std::cref(table->filter), std::cref(cb)
89+
)
10290
);
10391
}
104-
slave.setXidCallback(std::bind(&DBReader::XidEventCallback, this, _1, cb));
92+
slave.setXidCallback(std::bind(&DBReader::XidEventCallback, this, _1, std::cref(cb)));
10593
slave.init();
10694
slave.createDatabaseStructure();
10795

@@ -114,62 +102,106 @@ void DBReader::Stop()
114102
slave.close_connection();
115103
}
116104

117-
void DBReader::EventCallback(const slave::RecordSet& event, const std::map<std::string, unsigned>& filter, BinlogEventCallback cb)
118-
{
119-
last_event_when = event.when;
105+
void DBReader::EventCallbackNormal(
106+
const slave::RecordSet& event,
107+
const std::map<std::string, std::pair<unsigned, bool>>& filter,
108+
const BinlogEventCallback& cb
109+
) {
110+
if (stopped) return;
111+
// last_event_when = event.when;
120112

121113
SerializableBinlogEventPtr ev(new SerializableBinlogEvent);
114+
115+
switch (event.type_event) {
116+
case slave::RecordSet::Delete: ev->event = "DELETE"; break;
117+
case slave::RecordSet::Update: ev->event = "UPDATE"; break;
118+
case slave::RecordSet::Write: ev->event = "INSERT"; break;
119+
default: return;
120+
}
121+
122122
ev->binlog_name = state.getMasterLogName();
123123
ev->binlog_pos = state.getMasterLogPos();
124124
// ev->seconds_behind_master = GetSecondsBehindMaster();
125-
ev->unix_timestamp = long(time(NULL));
125+
// ev->unix_timestamp = long(time(NULL));
126126
ev->database = event.db_name;
127127
ev->table = event.tbl_name;
128128

129+
for (auto fi = filter.begin(), end = filter.end(); fi != end; ++fi) {
130+
const auto ri = event.m_row.find(fi->first);
131+
if (ri != event.m_row.end()) {
132+
ev->row[ fi->second.first ] = ri->second;
133+
}
134+
}
135+
stopped = cb(std::move(ev));
136+
}
137+
138+
void DBReader::EventCallbackNullify(
139+
const slave::RecordSet& event,
140+
const std::map<std::string, std::pair<unsigned, bool>>& filter,
141+
const BinlogEventCallback& cb
142+
) {
143+
if (stopped) return;
144+
// last_event_when = event.when;
145+
146+
SerializableBinlogEventPtr ev(new SerializableBinlogEvent);
147+
bool is_delete = false;
148+
129149
switch (event.type_event) {
130-
case slave::RecordSet::Delete: ev->event = "DELETE"; break;
150+
case slave::RecordSet::Delete: ev->event = "DELETE"; is_delete = true;
131151
case slave::RecordSet::Update: ev->event = "UPDATE"; break;
132152
case slave::RecordSet::Write: ev->event = "INSERT"; break;
133-
default: ev->event = "IGNORE";
153+
default: return;
134154
}
155+
156+
ev->binlog_name = state.getMasterLogName();
157+
ev->binlog_pos = state.getMasterLogPos();
158+
// ev->seconds_behind_master = GetSecondsBehindMaster();
159+
// ev->unix_timestamp = long(time(NULL));
160+
ev->database = event.db_name;
161+
ev->table = event.tbl_name;
162+
135163
for (auto fi = filter.begin(), end = filter.end(); fi != end; ++fi) {
136164
const auto ri = event.m_row.find(fi->first);
137165
if (ri != event.m_row.end()) {
138-
ev->row[ fi->second ] = ri->second;
166+
// if it's not a key and event is delete - don't actually delete, just nullify
167+
ev->row[ fi->second.first ] = !fi->second.second && is_delete ? boost::any() : ri->second;
139168
}
140169
}
141170
stopped = cb(std::move(ev));
142171
}
143172

144-
void DBReader::XidEventCallback(unsigned int server_id, BinlogEventCallback cb)
173+
void DBReader::XidEventCallback(unsigned int server_id, const BinlogEventCallback& cb)
145174
{
146-
last_event_when = ::time(NULL);
175+
if (stopped) return;
176+
// last_event_when = ::time(NULL);
147177

148178
// send binlog position update event
149179
SerializableBinlogEventPtr ev(new SerializableBinlogEvent);
150180
ev->binlog_name = state.getMasterLogName();
151181
ev->binlog_pos = state.getMasterLogPos();
152182
// ev->seconds_behind_master = GetSecondsBehindMaster();
153-
ev->unix_timestamp = long(time(NULL));
183+
// ev->unix_timestamp = long(time(NULL));
154184
ev->event = "IGNORE";
155185
stopped = cb(std::move(ev));
156186
}
157187

158188
void DBReader::DumpTablesCallback(
159-
const std::string &db_name,
160-
const std::string &tbl_name,
189+
const std::string& db_name,
190+
const std::string& tbl_name,
161191
const std::vector<std::pair<unsigned, slave::PtrField>>& filter,
162192
const nanomysql::fields_t& fields,
163-
BinlogEventCallback cb
193+
const BinlogEventCallback& cb
164194
) {
195+
if (stopped) return;
196+
165197
SerializableBinlogEventPtr ev(new SerializableBinlogEvent);
166198
ev->binlog_name = "";
167199
ev->binlog_pos = 0;
168200
ev->database = db_name;
169201
ev->table = tbl_name;
170202
ev->event = "INSERT";
171203
// ev->seconds_behind_master = GetSecondsBehindMaster();
172-
ev->unix_timestamp = long(time(NULL));
204+
// ev->unix_timestamp = long(time(NULL));
173205

174206
for (const auto& it : filter) {
175207
slave::PtrField ptr_field = it.second;
@@ -181,13 +213,11 @@ void DBReader::DumpTablesCallback(
181213
ev->row[ it.first ] = ptr_field->field_data;
182214
}
183215
}
184-
if (!stopped) {
185-
stopped = cb(std::move(ev));
186-
}
216+
stopped = cb(std::move(ev));
187217
}
188218

189-
unsigned DBReader::GetSecondsBehindMaster() const {
190-
return std::max(::time(NULL) - last_event_when, 0L);
191-
}
219+
// unsigned DBReader::GetSecondsBehindMaster() const {
220+
// return std::max(::time(NULL) - last_event_when, 0L);
221+
// }
192222

193223
} // replicator

‎dbreader.h

+47-24
Original file line numberDiff line numberDiff line change
@@ -18,48 +18,71 @@ typedef std::function<bool (SerializableBinlogEventPtr&&)> BinlogEventCallback;
1818

1919
struct DBTable
2020
{
21-
DBTable() {}
22-
23-
DBTable(const std::string& db, const std::string& table, std::map<std::string, unsigned>& filter, bool do_dump_)
24-
: name(db, table), filter(filter), do_dump(do_dump_) {}
25-
26-
std::pair<std::string, std::string> name;
27-
std::map<std::string, unsigned> filter;
28-
bool do_dump;
21+
DBTable(
22+
const std::string& db,
23+
const std::string& table,
24+
const std::map<std::string, std::pair<unsigned, bool>>& filter_,
25+
bool is_primary_
26+
) :
27+
name(db, table), filter(filter_), is_primary(is_primary_)
28+
{}
29+
30+
const std::pair<std::string, std::string> name;
31+
const std::map<std::string, std::pair<unsigned, bool>> filter;
32+
const bool is_primary;
2933
};
3034

3135
class DBReader
3236
{
3337
public:
34-
DBReader (nanomysql::mysql_conn_opts &opts, unsigned int connect_retry = 60);
35-
~DBReader();
36-
37-
void AddTable(const std::string &db, const std::string &table, std::map<std::string, unsigned>& filter, bool do_dump);
38-
void DumpTables(std::string &binlog_name, unsigned long &binlog_pos, BinlogEventCallback f);
39-
void ReadBinlog(const std::string &binlog_name, unsigned long binlog_pos, BinlogEventCallback cb);
38+
DBReader(nanomysql::mysql_conn_opts& opts, unsigned connect_retry)
39+
: state(), slave(slave::MasterInfo(opts, connect_retry), state), stopped(false) {}
40+
41+
~DBReader() {
42+
slave.close_connection();
43+
}
44+
45+
void AddTable(
46+
const std::string& db,
47+
const std::string& table,
48+
const std::map<std::string, std::pair<unsigned, bool>>& filter,
49+
bool is_primary
50+
) {
51+
tables.emplace_back(db, table, filter, is_primary);
52+
}
53+
54+
void DumpTables(std::string& binlog_name, unsigned long& binlog_pos, const BinlogEventCallback& cb);
55+
void ReadBinlog(const std::string& binlog_name, unsigned long binlog_pos, const BinlogEventCallback& cb);
4056
void Stop();
4157

42-
void EventCallback(const slave::RecordSet& event, const std::map<std::string, unsigned>& filter, BinlogEventCallback cb);
43-
void DummyEventCallback(const slave::RecordSet& event) {};
44-
45-
void XidEventCallback(unsigned int server_id, BinlogEventCallback cb);
58+
void EventCallbackNormal(
59+
const slave::RecordSet& event,
60+
const std::map<std::string, std::pair<unsigned, bool>>& filter,
61+
const BinlogEventCallback& cb
62+
);
63+
void EventCallbackNullify(
64+
const slave::RecordSet& event,
65+
const std::map<std::string, std::pair<unsigned, bool>>& filter,
66+
const BinlogEventCallback& cb
67+
);
68+
void XidEventCallback(unsigned int server_id, const BinlogEventCallback& cb);
4669
void DumpTablesCallback(
47-
const std::string &db_name,
48-
const std::string &tbl_name,
70+
const std::string& db_name,
71+
const std::string& tbl_name,
4972
const std::vector<std::pair<unsigned, slave::PtrField>>& filter,
50-
const nanomysql::fields_t &f,
51-
BinlogEventCallback cb
73+
const nanomysql::fields_t& f,
74+
const BinlogEventCallback& cb
5275
);
5376

54-
unsigned GetSecondsBehindMaster() const;
77+
// unsigned GetSecondsBehindMaster() const;
5578

5679
private:
5780
slave::DefaultExtState state;
5881
slave::Slave slave;
5982
std::vector<DBTable> tables;
6083
bool stopped;
6184

62-
::time_t last_event_when;
85+
// ::time_t last_event_when;
6386
};
6487

6588
} // replicator

‎main.cpp

+20-6
Original file line numberDiff line numberDiff line change
@@ -140,6 +140,7 @@ static void init(YAML::Node& cfg)
140140
}
141141
// read Mysql to Tarantool mappings (each table maps to a single Tarantool space)
142142
{
143+
std::map<unsigned, bool> has_primary;
143144
const YAML::Node& mappings = cfg["mappings"];
144145

145146
for (int i = 0; i < mappings.size(); i++) {
@@ -153,10 +154,17 @@ static void init(YAML::Node& cfg)
153154
std::string delete_call = mapping["delete_call"] ? mapping["delete_call"].as<std::string>() : TPWriter::empty_call;
154155

155156
const unsigned space = mapping["space"].as<unsigned>();
156-
std::map<std::string, unsigned> columns;
157-
TPWriter::Tuple keys;
157+
std::map<std::string, std::pair<unsigned, bool>> columns;
158+
std::vector<unsigned> keys;
158159
unsigned index_max = tpwriter->space_last_id[space];
159160

161+
bool is_primary;
162+
if (has_primary.find(space) == has_primary.end()) {
163+
is_primary = has_primary[space] = true;;
164+
} else {
165+
is_primary = false;
166+
}
167+
160168
// read key tarantool fields we'll use for delete requests
161169
{
162170
const YAML::Node& keys_ = mapping["key_fields"];
@@ -170,12 +178,18 @@ static void init(YAML::Node& cfg)
170178
{
171179
const YAML::Node& columns_ = mapping["columns"];
172180
for (int i = 0; i < columns_.size(); i++) {
173-
unsigned index = i < keys.size() ? keys[i] : ++index_max;
174-
columns[ columns_[i].as<std::string>() ] = index;
181+
const bool is_key = i < keys.size();
182+
const unsigned index = is_key ? keys[i] : ++index_max;
183+
columns.emplace(
184+
std::piecewise_construct,
185+
std::forward_as_tuple(columns_[i].as<std::string>()),
186+
std::forward_as_tuple(index, is_key)
187+
);
175188
}
176189
}
177190

178-
dbreader->AddTable(database, table, columns, mapping["dump"].as<bool>());
191+
dbreader->AddTable(database, table, columns, is_primary);
192+
std::sort(keys.begin(), keys.end());
179193
tpwriter->AddTable(database, table, space, keys, insert_call, update_call, delete_call);
180194
tpwriter->space_last_id[space] = index_max;
181195
}
@@ -194,7 +208,7 @@ static void init(YAML::Node& cfg)
194208
auto itf = field.begin();
195209
if (itf == field.end()) continue;
196210

197-
unsigned index = itrn->first.as<unsigned>();
211+
const unsigned index = itrn->first.as<unsigned>();
198212
std::string type = itf->first.as<std::string>();
199213
const YAML::Node& value = itf->second;
200214

0 commit comments

Comments
 (0)