SHS/pip
4
3
Fork 0
Code Issues 67 Pull Requests 1 Actions Releases Wiki Activity

binlog fixes

Browse Source 
PIBinaryStream doc
remove makePIPair
rename bytesAvailable
This commit is contained in:
Бычков Андрей
2022-07-28 14:46:58 +03:00
parent 00d06f71ba
commit 1b09ad5c27
19 changed files with 95 additions and 55 deletions
Download Patch File Download Diff File Expand all files Collapse all files

80
libs/main/io_devices/pibinarylog.cpp
View File

@@ -550,15 +550,33 @@ void PIBinaryLog::parseLog(PIFile * f, PIBinaryLog::BinLogInfo * info, PIVector<
}
uchar read_sig[PIBINARYLOG_SIGNATURE_SIZE];
for (uint i=0; i<PIBINARYLOG_SIGNATURE_SIZE; i++) read_sig[i] = 0;
bool ok = true;
if (f->read(read_sig, PIBINARYLOG_SIGNATURE_SIZE) < 0) {if (info) info->records_count = -1; ok = false;}
for (uint i=0; i<PIBINARYLOG_SIGNATURE_SIZE; i++)
if (read_sig[i] != binlog_sig[i]) {if (info) info->records_count = -2; ok = false;}
if (f->read(read_sig, PIBINARYLOG_SIGNATURE_SIZE) < 0) {
if (info) info->records_count = -1;
return;
}
for (uint i=0; i<PIBINARYLOG_SIGNATURE_SIZE; i++) {
if (read_sig[i] != binlog_sig[i]) {
if (info) info->records_count = -2;
return;
}
}
uchar read_version = 0;
if (f->read(&read_version, 1) < 0) {if (info) info->records_count = -3; ok = false;}
if (read_version == 0) {if (info) info->records_count = -4; ok = false;}
if (read_version < PIBINARYLOG_VERSION_OLD) {if (info) info->records_count = -5; ok = false;}
if (read_version > PIBINARYLOG_VERSION) {if (info) info->records_count = -6; ok = false;}
if (f->read(&read_version, 1) < 0) {
if (info) info->records_count = -3;
return;
}
if (read_version == 0) {
if (info) info->records_count = -4;
return;
}
if (read_version < PIBINARYLOG_VERSION_OLD) {
if (info) info->records_count = -5;
return;
}
if (read_version > PIBINARYLOG_VERSION) {
if (info) info->records_count = -6;
return;
}
if (read_version == PIBINARYLOG_VERSION) {
uint32_t sz = 0;
f->read(&sz, 4);
@@ -566,7 +584,6 @@ void PIBinaryLog::parseLog(PIFile * f, PIBinaryLog::BinLogInfo * info, PIVector<
info->user_header = f->read(sz);
}
}
if (!ok) return;
PIByteArray ba;
BinLogRecord br;
br.id = 0;
@@ -580,11 +597,15 @@ void PIBinaryLog::parseLog(PIFile * f, PIBinaryLog::BinLogInfo * info, PIVector<
if (f->read(ba.data(), ba.size()) > 0) {
ba >> br.id >> br.size >> br.timestamp;
} else break;
if (info->log_size - f->pos() >= br.size) f->seek(f->pos() + br.size);
if (info) {
if (info->log_size - f->pos() >= br.size) {
f->seek(f->pos() + br.size);
}
}
else break;
}
if (br.id > 0) {
if (info) {
if (index) {
BinLogIndex bl_ind;
bl_ind.id = br.id;
bl_ind.data_size = br.size;
@@ -674,15 +695,9 @@ bool PIBinaryLog::cutBinLog(const PIBinaryLog::BinLogInfo & src, const PIString
}
bool PIBinaryLog::joinBinLogsSerial(const PIStringList & src, const PIString & dst) {
bool PIBinaryLog::joinBinLogsSerial(const PIStringList & src, const PIString & dst, std::function<bool (const PIString &, PISystemTime)> progress) {
PIBinaryLog slog;
PIBinaryLog dlog;
dlog.createNewFile(dst);
if (!dlog.isOpened()) {
piCout << "[PIBinaryLog]" << "Error, can't create" << dst;
return false;
}
piCout << "[PIBinaryLog]" << "Start join binlogs to" << dst;
PISystemTime dtime;
PISystemTime lt;
PITimeMeasurer tm;
@@ -692,8 +707,16 @@ bool PIBinaryLog::joinBinLogsSerial(const PIStringList & src, const PIString & d
piCout << "[PIBinaryLog]" << "Error, can't open" << fn;
return false;
}
if (first) first = false;
else {
if (first) {
first = false;
dlog.setHeader(slog.getHeader());
dlog.createNewFile(dst);
if (!dlog.isOpened()) {
piCout << "[PIBinaryLog]" << "Error, can't create" << dst;
return false;
}
piCout << "[PIBinaryLog]" << "Start join binlogs to" << dst;
} else {
dtime = lt;
}
tm.reset();
@@ -701,18 +724,29 @@ bool PIBinaryLog::joinBinLogsSerial(const PIStringList & src, const PIString & d
PISystemTime st;
while (!slog.isEnd()) {
br = slog.readRecord();
if (br.data.isEmpty() || br.id < 1) continue;
st = br.timestamp;
lt = dtime + br.timestamp;
if (dlog.writeBinLog_raw(br.id, lt, br.data) <= 0) {
piCout << "[PIBinaryLog]" << "Error, can't write to file" << dst;
return false;
}
if (tm.elapsed_s() > 1) {
if (tm.elapsed_s() > 0.1) {
tm.reset();
piCout << "[PIBinaryLog]" << "process" << PITime::fromSystemTime(lt).toString();
if (progress) {
if (!progress(fn, lt)) {
slog.close();
dlog.close();
PIFile::remove(dlog.path());
return false;
}
} else {
piCout << "[PIBinaryLog]" << "process" << PITime::fromSystemTime(lt).toString();
}
}
}
piCout << "[PIBinaryLog]" << "complete" << fn;
slog.close();
//piCout << "[PIBinaryLog]" << "complete" << fn;
}
piCout << "[PIBinaryLog]" << "Finish join binlogs, total time" << PITime::fromSystemTime(lt).toString();
return true;