1004 lines
26 KiB
C++
1004 lines
26 KiB
C++
/*
|
|
PIP - Platform Independent Primitives
|
|
Class for write binary data to logfile, and read or playback this data
|
|
Andrey Bychkov work.a.b@yandex.ru
|
|
|
|
This program is free software: you can redistribute it and/or modify
|
|
it under the terms of the GNU Lesser General Public License as published by
|
|
the Free Software Foundation, either version 3 of the License, or
|
|
(at your option) any later version.
|
|
|
|
This program is distributed in the hope that it will be useful,
|
|
but WITHOUT ANY WARRANTY; without even the implied warranty of
|
|
MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
|
|
GNU Lesser General Public License for more details.
|
|
|
|
You should have received a copy of the GNU Lesser General Public License
|
|
along with this program. If not, see <http://www.gnu.org/licenses/>.
|
|
*/
|
|
|
|
#include "pibinarylog.h"
|
|
|
|
#include "pidir.h"
|
|
#include "pipropertystorage.h"
|
|
|
|
#define PIBINARYLOG_VERSION_OLD 0x31
|
|
|
|
/*! \class PIBinaryLog
|
|
* \brief Class for read and write binary data to logfile, and playback this data in realtime, or custom speed
|
|
*
|
|
* \section PIBinaryLog_sec0 Synopsis
|
|
* Binary Log is a file with simple header, where you can read and write some binary data.
|
|
* Any written data include special header with ID, size and timestamp.
|
|
* This header provides separation different messages from the one file by choosing different IDs.
|
|
* With \a filterID or special functions, like \a readBinLog() you can choose IDs what you want to read.
|
|
* With function \a writeBinLog() or \a setDefaultID() you can choose ID that mark you data.
|
|
* By default ID = 1, and \a filterID is empty, that mean you read any ID without filtering.
|
|
* ThreadedRead provide you playback data, with delay that you write data.
|
|
* You can choose different playbak modes by set \a PlayMode.
|
|
*
|
|
* \section PIBinaryLog_sec1 Basic usage
|
|
* This class provide all functions of \a PIIODevice, such \a open(), \a close(),
|
|
* \a read() ,\a write(), and threaded read/write.
|
|
* function \a setLogDir() need to set directory for BinLog files
|
|
* function \a createNewFile() need to create new binlog file
|
|
* function \a restart() need start from the begining of binlog file
|
|
*
|
|
*/
|
|
|
|
static const uchar binlog_sig[] = {'B', 'I', 'N', 'L', 'O', 'G'};
|
|
|
|
#define PIBINARYLOG_VERSION 0x32
|
|
#define PIBINARYLOG_SIGNATURE_SIZE sizeof(binlog_sig)
|
|
|
|
REGISTER_DEVICE(PIBinaryLog)
|
|
|
|
PIBinaryLog::PIBinaryLog() {
|
|
#ifdef MICRO_PIP
|
|
setThreadedReadBufferSize(512);
|
|
#else
|
|
setThreadedReadBufferSize(65536);
|
|
#endif
|
|
is_started = is_indexed = is_pause = false;
|
|
create_index_on_fly = false;
|
|
current_index = -1;
|
|
log_size = 0;
|
|
f_new_path = nullptr;
|
|
setPlaySpeed(1.);
|
|
setDefaultID(1);
|
|
setPlaySpeed(1.0);
|
|
setPlayDelay(PISystemTime::fromSeconds(1.0));
|
|
setPlayRealTime();
|
|
setSplitTime(PISystemTime(600, 0));
|
|
setSplitRecordCount(1000);
|
|
setSplitFileSize(0xFFFFFF);
|
|
setSplitMode(SplitNone);
|
|
setLogDir(PIString());
|
|
setFilePrefix(PIString());
|
|
setRapidStart(false);
|
|
file.setName("__S__PIBinaryLog::file");
|
|
}
|
|
|
|
|
|
PIBinaryLog::~PIBinaryLog() {
|
|
stop();
|
|
close();
|
|
}
|
|
|
|
|
|
bool PIBinaryLog::openDevice() {
|
|
lastrecord.timestamp = PISystemTime();
|
|
lastrecord.id = 0;
|
|
write_count = 0;
|
|
is_started = false;
|
|
is_thread_ok = true;
|
|
is_indexed = false;
|
|
is_pause = false;
|
|
index.clear();
|
|
log_size = 0;
|
|
if (mode_ == ReadWrite) {
|
|
piCoutObj << "Error: ReadWrite mode not supported, use WriteOnly or ReadOnly";
|
|
return false;
|
|
}
|
|
if (path().isEmpty() && mode_ == WriteOnly) {
|
|
if (f_new_path)
|
|
setPath(f_new_path());
|
|
else
|
|
setPath(getLogfilePath(logDir(), filePrefix()));
|
|
}
|
|
if (path().isEmpty() && mode_ == ReadOnly) {
|
|
PIDir ld(logDir());
|
|
if (ld.isExists()) {
|
|
PIVector<PIFile::FileInfo> es = ld.allEntries();
|
|
piForeachC(PIFile::FileInfo & i, es) {
|
|
if (i.extension() == "binlog" && i.isFile() && i.baseName().startsWith(filePrefix())) {
|
|
setPath(i.path);
|
|
break;
|
|
}
|
|
}
|
|
}
|
|
}
|
|
if (!file.open(path(), mode_)) {
|
|
piCoutObj << "Error: Can't open file" << path();
|
|
return false;
|
|
}
|
|
setName(path());
|
|
if (mode_ == WriteOnly) {
|
|
file.clear();
|
|
if (!writeFileHeader()) {
|
|
piCoutObj << "Error: Can't write binlog file header" << path();
|
|
return false;
|
|
}
|
|
is_started = true;
|
|
}
|
|
if (mode_ == ReadOnly) {
|
|
if (file.isEmpty()) {
|
|
piCoutObj << "Error: File is null" << path();
|
|
fileError();
|
|
return false;
|
|
}
|
|
if (!checkFileHeader()) {
|
|
fileError();
|
|
return false;
|
|
}
|
|
if (isEmpty()) {
|
|
piCoutObj << "Warning: Empty BinLog file" << path();
|
|
fileEnd();
|
|
}
|
|
play_time = 0;
|
|
if (!rapid_start) is_started = true;
|
|
}
|
|
startlogtime = PISystemTime::current();
|
|
pause_time = PISystemTime();
|
|
return true;
|
|
}
|
|
|
|
|
|
bool PIBinaryLog::closeDevice() {
|
|
stopAndWait();
|
|
pausemutex.unlock();
|
|
logmutex.unlock();
|
|
moveIndex(-1);
|
|
is_indexed = false;
|
|
index.clear();
|
|
bool e = isEmpty();
|
|
log_size = 0;
|
|
if (canWrite() && e) {
|
|
file.remove();
|
|
return true;
|
|
}
|
|
return file.close();
|
|
}
|
|
|
|
|
|
bool PIBinaryLog::threadedRead(const uchar * readed, ssize_t size) {
|
|
// piCout << "binlog threaded read";
|
|
if (!canRead() || isEnd()) return PIIODevice::threadedRead(readed, size);
|
|
is_thread_ok = false;
|
|
logmutex.lock();
|
|
PISystemTime pt, lastrec_timestamp = lastrecord.timestamp;
|
|
logmutex.unlock();
|
|
double delay;
|
|
switch (play_mode) {
|
|
case PlayRealTime:
|
|
pausemutex.lock();
|
|
if (is_pause) {
|
|
pausemutex.unlock();
|
|
piMSleep(100); // TODO: rewrite with condvar
|
|
return false;
|
|
} else if (pause_time > PISystemTime()) {
|
|
// startlogtime += pause_time;
|
|
pause_time = PISystemTime();
|
|
}
|
|
pausemutex.unlock();
|
|
pt = PISystemTime::current() - startlogtime;
|
|
if (is_started) {
|
|
if (lastrec_timestamp > pt) (lastrec_timestamp - pt).sleep();
|
|
} else {
|
|
startlogtime = PISystemTime::current() - lastrec_timestamp;
|
|
is_started = true;
|
|
}
|
|
break;
|
|
case PlayVariableSpeed:
|
|
delay = lastrec_timestamp.toMilliseconds() - play_time;
|
|
// piCoutObj << "delay" << delay;
|
|
double cdelay;
|
|
int dtc;
|
|
if (is_started) {
|
|
if (is_pause) {
|
|
piMSleep(100); // TODO: rewrite with condvar
|
|
return false;
|
|
}
|
|
if (delay > 0) {
|
|
cdelay = delay * play_speed; // TODO: rewrite with condvar
|
|
dtc = int(cdelay) / 100; // TODO: rewrite with condvar
|
|
if (play_speed <= 0.) dtc = 2;
|
|
// piCout << play_speed << dtc;
|
|
for (int j = 0; j < dtc; j++) {
|
|
cdelay = delay * play_speed; // TODO: rewrite with condvar
|
|
dtc = int(cdelay) / 100; // TODO: rewrite with condvar
|
|
piMSleep(100); // TODO: rewrite with condvar
|
|
if (play_speed <= 0.) {
|
|
dtc = 2;
|
|
j = 0;
|
|
}
|
|
// piCout << " " << play_speed << dtc << j;
|
|
}
|
|
cdelay = cdelay - dtc * 100; // TODO: rewrite with condvar
|
|
PISystemTime::fromMilliseconds(cdelay).sleep();
|
|
}
|
|
} else
|
|
is_started = true;
|
|
play_time = lastrec_timestamp.toMilliseconds();
|
|
break;
|
|
case PlayStaticDelay:
|
|
if (is_started) {
|
|
if (is_pause) {
|
|
piMSleep(100); // TODO: rewrite with condvar
|
|
return false;
|
|
}
|
|
play_delay.sleep();
|
|
} else
|
|
is_started = true;
|
|
break;
|
|
default: return false;
|
|
}
|
|
bool res = PIIODevice::threadedRead(readed, size);
|
|
is_thread_ok = true;
|
|
return res;
|
|
}
|
|
|
|
|
|
PIString PIBinaryLog::getLogfilePath(const PIString & log_dir, const PIString & prefix) {
|
|
PIDir dir(log_dir);
|
|
dir.setDir(dir.absolutePath());
|
|
if (!dir.isExists()) {
|
|
piCout << "[PIBinaryLog]"
|
|
<< "Creating directory" << dir.path();
|
|
dir.make(true);
|
|
}
|
|
PIString npath = log_dir + PIDir::separator + prefix + PIDateTime::current().toString("yyyy_MM_dd__hh_mm_ss");
|
|
PIString cnpath = npath + ".binlog";
|
|
int i = 1;
|
|
while (PIFile::isExists(cnpath)) {
|
|
cnpath = npath + '_' + PIString::fromNumber(i) + ".binlog";
|
|
i++;
|
|
}
|
|
return cnpath;
|
|
}
|
|
|
|
PIString PIBinaryLog::createNewFile() {
|
|
if (!file.close()) return PIString();
|
|
PIString cnpath;
|
|
if (f_new_path)
|
|
cnpath = f_new_path();
|
|
else
|
|
cnpath = getLogfilePath(logDir(), filePrefix());
|
|
if (open(cnpath, PIIODevice::WriteOnly)) {
|
|
newFile(file.path());
|
|
return file.path();
|
|
}
|
|
piCoutObj << "Can't create new file, maybe LogDir" << ("\"" + logDir() + "\"") << "is invalid.";
|
|
return PIString();
|
|
}
|
|
|
|
|
|
void PIBinaryLog::createNewFile(const PIString & path) {
|
|
if (open(path, PIIODevice::WriteOnly)) {
|
|
newFile(file.path());
|
|
} else
|
|
piCoutObj << "Can't create new file, maybe path" << ("\"" + path + "\"") << "is invalid.";
|
|
}
|
|
|
|
|
|
void PIBinaryLog::setCreateIndexOnFly(bool yes) {
|
|
create_index_on_fly = yes;
|
|
}
|
|
|
|
|
|
void PIBinaryLog::setPause(bool pause) {
|
|
pausemutex.lock();
|
|
is_pause = pause;
|
|
if (pause)
|
|
pause_time = PISystemTime::current();
|
|
else {
|
|
if (pause_time > PISystemTime()) pause_time = PISystemTime::current() - pause_time;
|
|
}
|
|
pausemutex.unlock();
|
|
}
|
|
|
|
|
|
int PIBinaryLog::writeBinLog(int id, const void * data, int size) {
|
|
if (size <= 0 || !canWrite()) return -1;
|
|
if (id == 0) {
|
|
piCoutObj << "Error: can`t write with id = 0! Id must be > 0";
|
|
return -1;
|
|
}
|
|
if (is_pause) return 0;
|
|
logmutex.lock();
|
|
int res = writeRecord(id, PISystemTime::current() - startlogtime, data, size);
|
|
switch (split_mode) {
|
|
case SplitSize:
|
|
if (log_size > split_size) createNewFile();
|
|
break;
|
|
case SplitTime:
|
|
if ((PISystemTime::current() - startlogtime) > split_time) createNewFile();
|
|
break;
|
|
case SplitCount:
|
|
if (write_count > split_count) createNewFile();
|
|
break;
|
|
default: break;
|
|
}
|
|
logmutex.unlock();
|
|
if (res > 0)
|
|
return size;
|
|
else
|
|
return res;
|
|
}
|
|
|
|
|
|
int PIBinaryLog::writeBinLog_raw(int id, const PISystemTime & time, const void * data, int size) {
|
|
if (size <= 0 || !canWrite()) return -1;
|
|
logmutex.lock();
|
|
int res = writeRecord(id, time, data, size);
|
|
logmutex.unlock();
|
|
if (res > 0)
|
|
return size;
|
|
else
|
|
return res;
|
|
}
|
|
|
|
|
|
PIByteArray PIBinaryLog::readBinLog(int id, PISystemTime * time, int * readed_id) {
|
|
if (!canRead()) return PIByteArray();
|
|
logmutex.lock();
|
|
Record br = readRecord();
|
|
logmutex.unlock();
|
|
if (br.id == -1) {
|
|
piCoutObj << "End of BinLog file";
|
|
fileEnd();
|
|
return PIByteArray();
|
|
}
|
|
if ((id == 0) && (br.id != 0) && (br.id != -1)) {
|
|
if (time) *time = br.timestamp;
|
|
if (readed_id) *readed_id = br.id;
|
|
return br.data;
|
|
}
|
|
logmutex.lock();
|
|
while (br.id != id && !isEnd())
|
|
br = readRecord();
|
|
logmutex.unlock();
|
|
if (br.id == -1) {
|
|
piCoutObj << "End of BinLog file";
|
|
fileEnd();
|
|
return PIByteArray();
|
|
}
|
|
if (br.id == id) {
|
|
if (time) *time = br.timestamp;
|
|
if (readed_id) *readed_id = br.id;
|
|
return br.data;
|
|
}
|
|
piCoutObj << "Can't find record with id =" << id;
|
|
return PIByteArray();
|
|
}
|
|
|
|
|
|
int PIBinaryLog::readBinLog(int id, void * read_to, int max_size, PISystemTime * time, int * readed_id) {
|
|
if (max_size <= 0 || read_to == 0) return -1;
|
|
PIByteArray ba = readBinLog(id, time, readed_id);
|
|
if (ba.isEmpty()) return -1;
|
|
int sz = piMini(max_size, ba.size());
|
|
memcpy(read_to, ba.data(), sz);
|
|
return sz;
|
|
}
|
|
|
|
|
|
bool PIBinaryLog::isEmpty() const {
|
|
return (log_size <= llong(PIBINARYLOG_SIGNATURE_SIZE + 1));
|
|
}
|
|
|
|
|
|
void PIBinaryLog::setHeader(const PIByteArray & header) {
|
|
user_header = header;
|
|
}
|
|
|
|
|
|
PIByteArray PIBinaryLog::getHeader() {
|
|
return index.info.user_header;
|
|
}
|
|
|
|
|
|
ssize_t PIBinaryLog::readDevice(void * read_to, ssize_t max_size) {
|
|
PIMutexLocker _ml(logmutex);
|
|
if (lastrecord.id == -1 || isEnd()) return 0;
|
|
if (!is_thread_ok && lastrecord.id > 0) return lastrecord.data.size();
|
|
if (!canRead()) return -1;
|
|
if (max_size <= 0 || read_to == 0) return -1;
|
|
Record br;
|
|
br.id = 0;
|
|
if (filterID.isEmpty()) {
|
|
br = readRecord();
|
|
} else {
|
|
while (!filterID.contains(br.id) && !isEnd())
|
|
br = readRecord();
|
|
}
|
|
if (br.id == -1) {
|
|
fileEnd();
|
|
piCoutObj << "End of BinLog file";
|
|
return 0;
|
|
}
|
|
if (br.id == 0) {
|
|
piCoutObj << "Read record error";
|
|
return -1;
|
|
}
|
|
ssize_t sz = piMini(max_size, br.data.size());
|
|
if (sz < br.data.size_s()) piCoutObj << "too small read buffer:" << max_size << ", data size:" << br.data.size();
|
|
memcpy(read_to, br.data.data(), sz);
|
|
return sz;
|
|
}
|
|
|
|
|
|
ssize_t PIBinaryLog::writeDevice(const void * data, ssize_t size) {
|
|
return writeBinLog(default_id, data, size);
|
|
}
|
|
|
|
|
|
void PIBinaryLog::restart() {
|
|
bool th = isThreadedRead();
|
|
if (th) stopThreadedRead();
|
|
if (!canRead()) return;
|
|
logmutex.unlock();
|
|
lastrecord.timestamp = PISystemTime();
|
|
lastrecord.id = 0;
|
|
is_thread_ok = true;
|
|
is_started = !rapidStart();
|
|
play_time = 0;
|
|
file.seekToBegin();
|
|
checkFileHeader();
|
|
moveIndex(0);
|
|
startlogtime = PISystemTime::current();
|
|
if (th) startThreadedRead();
|
|
}
|
|
|
|
|
|
bool PIBinaryLog::writeFileHeader() {
|
|
if (file.write(binlog_sig, PIBINARYLOG_SIGNATURE_SIZE) <= 0) return false;
|
|
uchar version = PIBINARYLOG_VERSION;
|
|
if (file.write(&version, 1) <= 0) return false;
|
|
uint32_t sz = user_header.size();
|
|
file.write(&sz, 4);
|
|
file.write(user_header);
|
|
file.flush();
|
|
if (create_index_on_fly) {
|
|
index.info.user_header = user_header;
|
|
index.info.path = file.path();
|
|
}
|
|
return true;
|
|
}
|
|
|
|
|
|
bool PIBinaryLog::checkFileHeader() {
|
|
index.info.user_header.clear();
|
|
uchar read_sig[PIBINARYLOG_SIGNATURE_SIZE];
|
|
for (uint i = 0; i < PIBINARYLOG_SIGNATURE_SIZE; i++)
|
|
read_sig[i] = 0;
|
|
if (file.read(read_sig, PIBINARYLOG_SIGNATURE_SIZE) < 0) return false;
|
|
bool correct = true;
|
|
for (uint i = 0; i < PIBINARYLOG_SIGNATURE_SIZE; i++)
|
|
if (read_sig[i] != binlog_sig[i]) correct = false;
|
|
if (!correct) {
|
|
piCoutObj << "BinLogFile signature is corrupted or invalid file";
|
|
return false;
|
|
}
|
|
uchar read_version = 0;
|
|
if (file.read(&read_version, 1) < 0) return false;
|
|
if (read_version == PIBINARYLOG_VERSION_OLD) {
|
|
log_size = file.size();
|
|
return true;
|
|
}
|
|
if (read_version == PIBINARYLOG_VERSION) {
|
|
log_size = file.size();
|
|
uint32_t sz = 0;
|
|
file.read(&sz, 4);
|
|
if (sz > 0) {
|
|
index.info.user_header = file.read(sz);
|
|
}
|
|
return true;
|
|
}
|
|
if (read_version == 0) piCoutObj << "BinLogFile has invalid version";
|
|
if (read_version < PIBINARYLOG_VERSION) piCoutObj << "BinLogFile has too old verion";
|
|
if (read_version > PIBINARYLOG_VERSION) piCoutObj << "BinLogFile has too newest version";
|
|
return false;
|
|
}
|
|
|
|
|
|
PIBinaryLog::Record PIBinaryLog::readRecord() {
|
|
// piCoutObj << "readRecord pos =" << file.pos();
|
|
logmutex.lock();
|
|
PIByteArray ba;
|
|
Record br;
|
|
lastrecord.id = 0;
|
|
lastrecord.data.clear();
|
|
lastrecord.timestamp = PISystemTime();
|
|
ba.resize(sizeof(Record) - sizeof(PIByteArray));
|
|
if (file.read(ba.data(), ba.size_s()) > 0) {
|
|
ba >> br.id >> br.size >> br.timestamp;
|
|
} else {
|
|
br.id = -1;
|
|
logmutex.unlock();
|
|
// piCoutObj << "readRecord done" << br.id;
|
|
return br;
|
|
}
|
|
if ((br.id != 0) && (br.id != -1) && (br.size > 0)) {
|
|
ba.resize(br.size);
|
|
if (file.read(ba.data(), ba.size_s()) > 0)
|
|
br.data = ba;
|
|
else
|
|
br.id = 0;
|
|
} else
|
|
br.id = 0;
|
|
lastrecord = br;
|
|
if (br.id == 0) fileError();
|
|
moveIndex(index.index_pos.value(file.pos(), -1));
|
|
logmutex.unlock();
|
|
// piCoutObj << "readRecord done";
|
|
return br;
|
|
}
|
|
|
|
|
|
int PIBinaryLog::writeRecord(int id, PISystemTime time, const void * data, int size) {
|
|
PIByteArray logdata;
|
|
logdata << id << size << time << PIMemoryBlock(data, size);
|
|
if (create_index_on_fly) {
|
|
BinLogIndex bi;
|
|
bi.id = id;
|
|
bi.data_size = size;
|
|
bi.pos = file.pos();
|
|
bi.timestamp = time;
|
|
index.index << bi;
|
|
auto & rinfo(index.info.records[id]);
|
|
if (rinfo.count == 0) {
|
|
rinfo.id = id;
|
|
rinfo.start_time = rinfo.end_time = time;
|
|
rinfo.minimum_size = rinfo.maximum_size = size;
|
|
} else {
|
|
rinfo.start_time = piMin(rinfo.start_time, time);
|
|
rinfo.end_time = piMax(rinfo.end_time, time);
|
|
rinfo.minimum_size = piMin(rinfo.minimum_size, size);
|
|
rinfo.maximum_size = piMax(rinfo.maximum_size, size);
|
|
}
|
|
++rinfo.count;
|
|
if (index.info.records_count == 0) {
|
|
index.info.start_time = index.info.end_time = time;
|
|
} else {
|
|
index.info.start_time = piMin(index.info.start_time, time);
|
|
index.info.end_time = piMax(index.info.end_time, time);
|
|
}
|
|
++index.info.records_count;
|
|
is_indexed = true;
|
|
}
|
|
int ret = file.write(logdata.data(), logdata.size());
|
|
file.flush();
|
|
write_count++;
|
|
log_size = file.size();
|
|
if (create_index_on_fly) {
|
|
index.info.log_size = log_size;
|
|
}
|
|
return ret;
|
|
}
|
|
|
|
|
|
void PIBinaryLog::parseLog(PIFile * f, PIBinaryLog::BinLogInfo * info, PIVector<PIBinaryLog::BinLogIndex> * index) {
|
|
if (!info && !index) return;
|
|
if (info) {
|
|
info->log_size = -1;
|
|
info->records_count = 0;
|
|
info->records.clear();
|
|
}
|
|
if (index) index->clear();
|
|
if (f == 0) return;
|
|
if (!f->canRead()) return;
|
|
if (info) {
|
|
info->path = f->path();
|
|
info->log_size = f->size();
|
|
}
|
|
uchar read_sig[PIBINARYLOG_SIGNATURE_SIZE];
|
|
for (uint i = 0; i < PIBINARYLOG_SIGNATURE_SIZE; i++)
|
|
read_sig[i] = 0;
|
|
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;
|
|
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);
|
|
if (sz > 0) {
|
|
PIByteArray user_hdr = f->read(sz);
|
|
if (info) info->user_header = user_hdr;
|
|
}
|
|
}
|
|
PIByteArray ba;
|
|
Record br;
|
|
br.id = 0;
|
|
br.size = 0;
|
|
bool first = true;
|
|
size_t hdr_size = sizeof(Record) - sizeof(PIByteArray);
|
|
ba.resize(hdr_size);
|
|
while (1) {
|
|
ba.resize(hdr_size);
|
|
{
|
|
if (f->read(ba.data(), ba.size()) > 0) {
|
|
ba >> br.id >> br.size >> br.timestamp;
|
|
} else
|
|
break;
|
|
if (info) {
|
|
if (info->log_size - f->pos() >= br.size) {
|
|
f->seek(f->pos() + br.size);
|
|
}
|
|
} else
|
|
break;
|
|
}
|
|
if (br.id > 0) {
|
|
if (index) {
|
|
BinLogIndex bl_ind;
|
|
bl_ind.id = br.id;
|
|
bl_ind.data_size = br.size;
|
|
bl_ind.pos = f->pos() - br.size - hdr_size;
|
|
bl_ind.timestamp = br.timestamp;
|
|
index->append(bl_ind);
|
|
}
|
|
if (info) {
|
|
info->records_count++;
|
|
if (first) {
|
|
info->start_time = br.timestamp;
|
|
first = false;
|
|
}
|
|
BinLogRecordInfo & bri(info->records[br.id]);
|
|
bri.count++;
|
|
if (bri.id == 0) {
|
|
bri.id = br.id;
|
|
bri.minimum_size = bri.maximum_size = br.size;
|
|
bri.start_time = br.timestamp;
|
|
} else {
|
|
bri.end_time = br.timestamp;
|
|
if (bri.minimum_size > br.size) bri.minimum_size = br.size;
|
|
if (bri.maximum_size < br.size) bri.maximum_size = br.size;
|
|
}
|
|
}
|
|
}
|
|
}
|
|
if (info) info->end_time = br.timestamp;
|
|
}
|
|
|
|
|
|
void PIBinaryLog::moveIndex(int i) {
|
|
if (is_indexed) {
|
|
current_index = i;
|
|
posChanged(current_index);
|
|
}
|
|
}
|
|
|
|
|
|
PIBinaryLog::BinLogInfo PIBinaryLog::getLogInfo(const PIString & path) {
|
|
BinLogInfo bi;
|
|
bi.path = path;
|
|
bi.records_count = 0;
|
|
PIFile tfile;
|
|
if (!tfile.open(path, PIIODevice::ReadOnly)) return bi;
|
|
parseLog(&tfile, &bi, 0);
|
|
return bi;
|
|
}
|
|
|
|
|
|
bool PIBinaryLog::cutBinLog(const PIBinaryLog::BinLogInfo & src, const PIString & dst, int from, int to) {
|
|
PIBinaryLog slog;
|
|
if (!slog.open(src.path, PIIODevice::ReadOnly)) {
|
|
piCout << "[PIBinaryLog]"
|
|
<< "Error, can't open" << src.path;
|
|
return false;
|
|
}
|
|
PIVector<int> ids = src.records.keys();
|
|
slog.seekTo(from);
|
|
PIBinaryLog dlog;
|
|
dlog.createNewFile(dst);
|
|
if (!dlog.isOpened()) {
|
|
piCout << "[PIBinaryLog]"
|
|
<< "Error, can't create" << dst;
|
|
return false;
|
|
}
|
|
bool first = true;
|
|
Record br;
|
|
PISystemTime st;
|
|
PITimeMeasurer tm;
|
|
while (!slog.isEnd() && ((slog.pos() <= to) || to < 0)) {
|
|
br = slog.readRecord();
|
|
if (first) {
|
|
st = br.timestamp;
|
|
first = false;
|
|
}
|
|
if (ids.contains(br.id)) {
|
|
if (dlog.writeBinLog_raw(br.id, br.timestamp - st, br.data) <= 0) {
|
|
piCout << "[PIBinaryLog]"
|
|
<< "Error, can't write to file" << dst;
|
|
return false;
|
|
}
|
|
}
|
|
if (tm.elapsed_s() > 1) {
|
|
tm.reset();
|
|
piCout << "[PIBinaryLog]"
|
|
<< "process" << PITime::fromSystemTime(br.timestamp).toString();
|
|
}
|
|
}
|
|
return true;
|
|
}
|
|
|
|
|
|
bool PIBinaryLog::joinBinLogsSerial(const PIStringList & src,
|
|
const PIString & dst,
|
|
std::function<bool(const PIString &, PISystemTime)> progress) {
|
|
PIBinaryLog slog;
|
|
PIBinaryLog dlog;
|
|
PISystemTime dtime;
|
|
PISystemTime lt;
|
|
PITimeMeasurer tm;
|
|
bool first = true;
|
|
for (const PIString & fn: src) {
|
|
if (!slog.open(fn, PIIODevice::ReadOnly)) {
|
|
piCout << "[PIBinaryLog]"
|
|
<< "Error, can't open" << fn;
|
|
return false;
|
|
}
|
|
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();
|
|
Record br;
|
|
PISystemTime st;
|
|
while (!slog.isEnd()) {
|
|
br = slog.readRecord();
|
|
if (br.data.isEmpty() || br.id == 0 || 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() > 0.1) {
|
|
tm.reset();
|
|
if (progress) {
|
|
if (!progress(fn, lt)) {
|
|
slog.close();
|
|
dlog.close();
|
|
PIFile::remove(dlog.path());
|
|
return false;
|
|
}
|
|
} else {
|
|
piCout << "[PIBinaryLog]"
|
|
<< "process" << PITime::fromSystemTime(lt).toString();
|
|
}
|
|
}
|
|
}
|
|
slog.close();
|
|
// piCout << "[PIBinaryLog]" << "complete" << fn;
|
|
}
|
|
piCout << "[PIBinaryLog]"
|
|
<< "Finish join binlogs, total time" << PITime::fromSystemTime(lt).toString();
|
|
return true;
|
|
}
|
|
|
|
|
|
bool PIBinaryLog::createIndex() {
|
|
logmutex.lock();
|
|
llong cp = file.pos();
|
|
file.seekToBegin();
|
|
index.clear();
|
|
parseLog(&file, &index.info, &index.index);
|
|
index.makeIndexPos();
|
|
file.seek(cp);
|
|
is_indexed = !index.index.isEmpty();
|
|
logmutex.unlock();
|
|
return is_indexed;
|
|
}
|
|
|
|
|
|
int PIBinaryLog::posForTime(const PISystemTime & time) {
|
|
int ci = -1;
|
|
for (uint i = 0; i < index.index.size(); i++) {
|
|
if (time <= index.index[i].timestamp && (filterID.contains(index.index[i].id) || filterID.isEmpty())) {
|
|
ci = i;
|
|
break;
|
|
}
|
|
}
|
|
return ci;
|
|
}
|
|
|
|
|
|
void PIBinaryLog::seekTo(int rindex) {
|
|
// piCoutObj << "seekTo";
|
|
logmutex.lock();
|
|
pausemutex.lock();
|
|
if (rindex < index.index.size_s() && rindex >= 0) {
|
|
file.seek(index.index[rindex].pos);
|
|
moveIndex(index.index_pos.value(file.pos(), -1));
|
|
// double prev_pt = play_time;
|
|
play_time = index.index[rindex].timestamp.toMilliseconds();
|
|
lastrecord.timestamp = index.index[rindex].timestamp;
|
|
if (play_mode == PlayRealTime) {
|
|
startlogtime = PISystemTime::current() - lastrecord.timestamp;
|
|
}
|
|
}
|
|
// piCoutObj << "seekTo done";
|
|
pausemutex.unlock();
|
|
logmutex.unlock();
|
|
}
|
|
|
|
|
|
bool PIBinaryLog::seek(const PISystemTime & time) {
|
|
int ci = posForTime(time);
|
|
if (ci >= 0) {
|
|
seekTo(ci);
|
|
return true;
|
|
}
|
|
return false;
|
|
}
|
|
|
|
|
|
bool PIBinaryLog::seek(llong filepos) {
|
|
int ci = -1;
|
|
for (uint i = 0; i < index.index.size(); i++) {
|
|
if (filepos <= index.index[i].pos && (filterID.contains(index.index[i].id) || filterID.isEmpty())) {
|
|
ci = i;
|
|
break;
|
|
}
|
|
}
|
|
if (ci >= 0) {
|
|
seekTo(ci);
|
|
return true;
|
|
}
|
|
return false;
|
|
}
|
|
|
|
|
|
int PIBinaryLog::pos() const {
|
|
if (is_indexed) return current_index;
|
|
return -1;
|
|
}
|
|
|
|
|
|
PIByteArray PIBinaryLog::saveIndex() const {
|
|
if (!is_indexed) return PIByteArray();
|
|
return piSerialize(index);
|
|
}
|
|
|
|
|
|
bool PIBinaryLog::loadIndex(PIByteArray saved) {
|
|
if (!canRead() || saved.isEmpty()) return false;
|
|
index = piDeserialize<CompleteIndex>(saved);
|
|
index.makeIndexPos();
|
|
is_indexed = index.index.isNotEmpty();
|
|
return is_indexed;
|
|
}
|
|
|
|
|
|
PIString PIBinaryLog::constructFullPathDevice() const {
|
|
PIString ret;
|
|
ret += logDir() + ":" + filePrefix() + ":" + PIString::fromNumber(defaultID()) + ":";
|
|
switch (play_mode) {
|
|
case PlayRealTime: ret += "RT"; break;
|
|
case PlayVariableSpeed: ret += PIString::fromNumber(playSpeed()) + "X"; break;
|
|
case PlayStaticDelay: ret += PIString::fromNumber(playDelay().toMilliseconds()) + "M"; break;
|
|
default: ret += "RT"; break;
|
|
}
|
|
return ret;
|
|
}
|
|
|
|
|
|
void PIBinaryLog::configureFromFullPathDevice(const PIString & full_path) {
|
|
PIStringList pl = full_path.split(":");
|
|
for (int i = 0; i < pl.size_s(); ++i) {
|
|
PIString p(pl[i]);
|
|
switch (i) {
|
|
case 0: setLogDir(p); break;
|
|
case 1: setFilePrefix(p); break;
|
|
case 2: setDefaultID(p.toInt()); break;
|
|
case 3:
|
|
if (p.toUpperCase() == "RT") setPlayRealTime();
|
|
if (p.toUpperCase().right(1) == "X") setPlaySpeed((p.left(p.size() - 1)).toDouble());
|
|
if (p.toUpperCase().right(1) == "M") setPlayDelay(PISystemTime::fromMilliseconds((p.left(p.size() - 1)).toDouble()));
|
|
break;
|
|
}
|
|
}
|
|
// piCoutObj << "configured";
|
|
}
|
|
|
|
|
|
PIPropertyStorage PIBinaryLog::constructVariantDevice() const {
|
|
PIPropertyStorage ret;
|
|
PIVariantTypes::Enum e;
|
|
ret.addProperty("log dir", PIVariantTypes::Dir(logDir()));
|
|
ret.addProperty("file prefix", filePrefix());
|
|
ret.addProperty("default ID", defaultID());
|
|
e << "real-time"
|
|
<< "variable speed"
|
|
<< "static delay";
|
|
e.selectValue((int)playMode());
|
|
ret.addProperty("play mode", e);
|
|
ret.addProperty("play speed", playSpeed());
|
|
ret.addProperty("play delay", playDelay().toMilliseconds());
|
|
return ret;
|
|
}
|
|
|
|
|
|
void PIBinaryLog::configureFromVariantDevice(const PIPropertyStorage & d) {
|
|
setLogDir(d.propertyValueByName("log dir").toString());
|
|
setFilePrefix(d.propertyValueByName("file prefix").toString());
|
|
setDefaultID(d.propertyValueByName("default ID").toInt());
|
|
setPlaySpeed(d.propertyValueByName("play speed").toDouble());
|
|
setPlayDelay(PISystemTime::fromMilliseconds(d.propertyValueByName("play delay").toDouble()));
|
|
setPlayMode((PlayMode)d.propertyValueByName("play mode").toEnum().selectedValue());
|
|
}
|
|
|
|
|
|
void PIBinaryLog::propertyChanged(const char * s) {
|
|
default_id = property("defaultID").toInt();
|
|
rapid_start = property("rapidStart").toBool();
|
|
play_mode = (PlayMode)property("playMode").toInt();
|
|
double ps = property("playSpeed").toDouble();
|
|
play_speed = ps > 0. ? 1. / ps : 0.;
|
|
play_delay = property("playDelay").toSystemTime();
|
|
split_mode = (SplitMode)property("splitMode").toInt();
|
|
split_time = property("splitTime").toSystemTime();
|
|
split_size = property("splitFileSize").toLLong();
|
|
split_count = property("splitRecordCount").toInt();
|
|
// piCoutObj << "propertyChanged" << s << play_mode;
|
|
}
|
|
|
|
|
|
void PIBinaryLog::CompleteIndex::clear() {
|
|
info = BinLogInfo();
|
|
index.clear();
|
|
index_pos.clear();
|
|
}
|
|
|
|
|
|
void PIBinaryLog::CompleteIndex::makeIndexPos() {
|
|
index_pos.clear();
|
|
for (uint i = 0; i < index.size(); i++)
|
|
index_pos[index[i].pos] = i;
|
|
}
|