new PIBaseTransfer, PIDataTransfer and update PIFileTransfer
git-svn-id: svn://db.shs.com.ru/pip@9 12ceb7fc-bf1f-11e4-8940-5bc7170c53b5
This commit is contained in:
@@ -1,374 +1,24 @@
|
||||
#include "pidatatransfer.h"
|
||||
|
||||
const uint PIBaseTransfer::signature = 0x54444950;
|
||||
|
||||
PIBaseTransfer::PIBaseTransfer(): crc(standardCRC_16()) {
|
||||
header.sig = signature;
|
||||
header.session_id = 0;
|
||||
packet_header_size = sizeof(PacketHeader);
|
||||
part_header_size = sizeof(Part) + sizeof(PIByteArray);
|
||||
is_sending = is_receiving = false;
|
||||
break_ = true;
|
||||
bytes_all = bytes_cur = 0;
|
||||
timeout_ = 1.;
|
||||
setPacketSize(4096);
|
||||
srand(PISystemTime::current().toMilliseconds());
|
||||
}
|
||||
|
||||
|
||||
PIBaseTransfer::~PIBaseTransfer() {
|
||||
break_ = true;
|
||||
session.clear();
|
||||
replies.clear();
|
||||
}
|
||||
|
||||
|
||||
void PIBaseTransfer::stopSend() {
|
||||
if (!is_sending) return;
|
||||
break_ = true;
|
||||
}
|
||||
|
||||
|
||||
void PIBaseTransfer::stopReceive() {
|
||||
if (!is_receiving) return;
|
||||
break_ = true;
|
||||
finish_receive(false);
|
||||
}
|
||||
|
||||
|
||||
void PIBaseTransfer::received(PIByteArray& data) {
|
||||
if (data.size() < sizeof(PacketHeader)) return;
|
||||
PacketHeader h;
|
||||
memcpy(&h, data.data(), sizeof(PacketHeader));
|
||||
PacketType pt = (PacketType)h.type;
|
||||
// piCoutObj << "receive" << h.session_id << h.type << h.id;
|
||||
switch (pt) {
|
||||
case pt_Unknown:
|
||||
break;
|
||||
case pt_Data:
|
||||
if (h.session_id != header.session_id || !is_receiving) {
|
||||
sendBreak(h.session_id);
|
||||
return;
|
||||
} else {
|
||||
uint rcrc = h.crc;
|
||||
uint ccrc = crc.calculate(data.data(sizeof(PacketHeader)), data.size_s() - sizeof(PacketHeader));
|
||||
if (rcrc != ccrc) {
|
||||
header.id = h.id;
|
||||
sendReply(pt_ReplyInvalid);
|
||||
} else {
|
||||
data >> h;
|
||||
processData(h.id, data);
|
||||
}
|
||||
}
|
||||
break;
|
||||
case pt_ReplySuccess:
|
||||
case pt_ReplyInvalid:
|
||||
if (h.session_id != header.session_id) return;
|
||||
if (is_sending) {
|
||||
if (h.id >= 0 && h.id < replies.size())
|
||||
replies[h.id] = pt;
|
||||
}
|
||||
if (is_receiving && h.id == 0) {
|
||||
if (checkSession() == 0 && pt == pt_ReplySuccess) finish_receive(true);
|
||||
}
|
||||
break;
|
||||
case pt_Break:
|
||||
break_ = true;
|
||||
if (is_receiving) {
|
||||
stopReceive();
|
||||
return;
|
||||
}
|
||||
if (is_sending) {
|
||||
stopSend();
|
||||
return;
|
||||
}
|
||||
break;
|
||||
case pt_Start:
|
||||
if (is_sending) {
|
||||
sendBreak(h.session_id);
|
||||
return;
|
||||
}
|
||||
if (header.session_id != h.session_id && is_receiving) {
|
||||
sendBreak(h.session_id);
|
||||
return;
|
||||
}
|
||||
if (data.size() == sizeof(StartRequest) + sizeof(PacketHeader)) {
|
||||
StartRequest sr;
|
||||
memcpy(&sr, data.data(sizeof(PacketHeader)), sizeof(StartRequest));
|
||||
bytes_all = sr.size;
|
||||
header.session_id = h.session_id;
|
||||
header.id = 0;
|
||||
state_string = "start request";
|
||||
session.clear();
|
||||
replies.clear();
|
||||
session.resize(sr.packets);
|
||||
replies.resize(sr.packets + 1);
|
||||
replies.fill(pt_Unknown);
|
||||
is_receiving = true;
|
||||
break_ = false;
|
||||
state_string = "receiving";
|
||||
replies[0] = pt_ReplySuccess;
|
||||
sendReply(pt_ReplySuccess);
|
||||
}
|
||||
break;
|
||||
default:
|
||||
break;
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
bool PIBaseTransfer::send_process() {
|
||||
break_ = false;
|
||||
is_sending = true;
|
||||
startSend();
|
||||
replies.resize(session.size() + 1);
|
||||
replies.fill(pt_Unknown);
|
||||
PIByteArray PIDataTransfer::buildPacket(Part fi) {
|
||||
PIByteArray ba;
|
||||
if (!getStartRequest()) return finish_send(false);
|
||||
for (int i = 0; i < session.size_s(); i++) {
|
||||
ba = build_packet(i);
|
||||
sendRequest(ba);
|
||||
if (break_) return finish_send(false);
|
||||
}
|
||||
// piCoutObj << "correct errors";
|
||||
PITimeMeasurer tm;
|
||||
int prev_chk = 0;
|
||||
while (tm.elapsed_s() < timeout_) {
|
||||
int chk = checkSession();
|
||||
if (chk != prev_chk) tm.reset();
|
||||
if (chk == 0) return finish_send(true);
|
||||
if (chk > 0) {
|
||||
ba = buildPacket(chk - 1);
|
||||
sendRequest(ba);
|
||||
}
|
||||
// if (chk == -1) return finish_send(false);
|
||||
if (break_) return finish_send(false);
|
||||
prev_chk = chk;
|
||||
piMSleep(1);
|
||||
}
|
||||
return finish_send(false);
|
||||
ba.resize(fi.size);
|
||||
memcpy(ba.data(), data_.data(fi.start), fi.size);
|
||||
return ba;
|
||||
}
|
||||
|
||||
|
||||
int PIBaseTransfer::checkSession() {
|
||||
int miss = 0;
|
||||
for (int i = 1; i < replies.size_s(); i++) {
|
||||
if (replies[i] != pt_ReplySuccess) miss++;
|
||||
if (replies[i] == pt_ReplyInvalid) return i;
|
||||
}
|
||||
for (int i = 1; i < replies.size_s(); i++) {
|
||||
if (replies[i] != pt_ReplySuccess) return i;
|
||||
}
|
||||
if (miss > 0) {
|
||||
piCoutObj << "missing" << miss << "packets";
|
||||
return -miss;
|
||||
} else return 0;
|
||||
void PIDataTransfer::receivePart(Part fi, PIByteArray ba, PIByteArray pheader) {
|
||||
if (data_.size() < fi.start + fi.size) data_.resize(fi.start + fi.size);
|
||||
memcpy(data_.data(fi.start), ba.data(), ba.size_s());
|
||||
}
|
||||
|
||||
|
||||
void PIBaseTransfer::buildSession(PIVector<Part> parts) {
|
||||
state_string = "calculating files... ";
|
||||
session.clear();
|
||||
header.session_id = rand();
|
||||
bytes_all = 0;
|
||||
Part fi;
|
||||
int fi_index, fi_prts;
|
||||
PIVector<Part> lfi;
|
||||
int min_size = packet_header_size + part_header_size;
|
||||
int cur_size = min_size;
|
||||
for (int i = 0; i < parts.size_s(); i++) {
|
||||
state_string = "calculating files... " + PIString::fromNumber(i) + " of " + PIString::fromNumber(parts.size());
|
||||
fi.id = parts[i].id;
|
||||
bytes_all += fi.size;
|
||||
// fi.size = fi.entry.size;
|
||||
fi.start = 0;
|
||||
int rest = parts[i].size - (packet_size - cur_size);
|
||||
// piCout << i << fi.entry << rest;
|
||||
if (rest <= 0) {
|
||||
fi.size = parts[i].size;
|
||||
lfi << fi;
|
||||
cur_size += fi.size + part_header_size;
|
||||
} else {
|
||||
fi.size = parts[i].size - rest;
|
||||
fi_index = 1;
|
||||
fi_prts = 1 + 1 + piMaxi(1, rest / (packet_size - min_size));
|
||||
// piCout << fi_prts;
|
||||
lfi << fi;
|
||||
session << lfi;
|
||||
lfi.clear();
|
||||
cur_size = min_size;
|
||||
llong fs = fi.size;
|
||||
for (int j = 1; j < fi_prts; j++) {
|
||||
fi_index++;
|
||||
fi.start = fs;
|
||||
fi.size = piMin<ullong>(parts[i].size - fs, packet_size - min_size);
|
||||
lfi << fi;
|
||||
cur_size += fi.size + part_header_size;
|
||||
if (fi_index != fi_prts) {
|
||||
session << lfi;
|
||||
lfi.clear();
|
||||
cur_size = min_size;
|
||||
fs += fi.size;
|
||||
}
|
||||
}
|
||||
}
|
||||
if (packet_size - cur_size < min_size) {
|
||||
session << lfi;
|
||||
lfi.clear();
|
||||
cur_size = min_size;
|
||||
}
|
||||
}
|
||||
if (cur_size > min_size) session << lfi;
|
||||
bool PIDataTransfer::send(const PIByteArray& ba) {
|
||||
data_ = ba;
|
||||
buildSession(PIVector<Part>() << Part(0, data_.size()));
|
||||
return send_process();
|
||||
}
|
||||
|
||||
|
||||
void PIBaseTransfer::sendBreak(int session_id) {
|
||||
uint psid = header.session_id;
|
||||
header.session_id = session_id;
|
||||
sendReply(pt_Break);
|
||||
header.session_id = psid;
|
||||
}
|
||||
|
||||
|
||||
void PIBaseTransfer::sendReply(PacketType reply) {
|
||||
header.type = reply;
|
||||
PIByteArray ba;
|
||||
ba << header;
|
||||
sendRequest(ba);
|
||||
}
|
||||
|
||||
|
||||
bool PIBaseTransfer::getStartRequest() {
|
||||
PITimeMeasurer tm;
|
||||
header.type = pt_Start;
|
||||
header.id = 0;
|
||||
PIByteArray ba;
|
||||
ba << header;
|
||||
ba << (uint)session.size() << bytes_all;
|
||||
state_string = "send request";
|
||||
for (int i = 0; i < 3; i++) {
|
||||
tm.reset();
|
||||
sendRequest(ba);
|
||||
while (tm.elapsed_s() < timeout_) {
|
||||
if (break_) return false;
|
||||
//piCoutObj << send_replyes[0];
|
||||
if (replies[0] == pt_ReplySuccess) {
|
||||
state_string = "send permited!";
|
||||
return true;
|
||||
}
|
||||
piMSleep(10);
|
||||
}
|
||||
}
|
||||
return false;
|
||||
}
|
||||
|
||||
|
||||
void PIBaseTransfer::processData(int id, PIByteArray & data) {
|
||||
// piCoutObj << "received packet" << id << ", size" << data.size();
|
||||
if (id < 1 || id > replies.size_s()) return;
|
||||
if (!session[id - 1].isEmpty()) {
|
||||
header.id = id;
|
||||
replies[id] = pt_ReplySuccess;
|
||||
sendReply(pt_ReplySuccess);
|
||||
if (checkSession() == 0) state_string = "receive ok";
|
||||
return;
|
||||
}
|
||||
Part fi;
|
||||
PIByteArray ba, pheader;
|
||||
pheader.resize(packet_header_size - sizeof(PacketHeader));
|
||||
if (!pheader.isEmpty()) {
|
||||
memcpy(pheader.data(), data.data(), pheader.size());
|
||||
data.remove(0, pheader.size_s());
|
||||
}
|
||||
while (!data.isEmpty()) {
|
||||
ba.clear();
|
||||
data >> fi;
|
||||
if (fi.size > 0) data >> ba;
|
||||
// fi.fsize = ba.size();
|
||||
bytes_cur += fi.size;
|
||||
// piCoutObj << "recv" << fi;
|
||||
session[id - 1] << fi;
|
||||
state_string = "receiving...";
|
||||
receivePart(fi, ba, pheader);
|
||||
}
|
||||
header.id = id;
|
||||
replies[id] = pt_ReplySuccess;
|
||||
if (checkSession() == 0) state_string = "receive ok";
|
||||
sendReply(pt_ReplySuccess);
|
||||
}
|
||||
|
||||
|
||||
PIByteArray PIBaseTransfer::build_packet(int id) {
|
||||
PIByteArray ret;
|
||||
PIByteArray ba;
|
||||
header.id = id + 1;
|
||||
header.type = pt_Data;
|
||||
//piCoutObj << "Packet" << header.id;
|
||||
//piCoutObj << "session id" << header.session_id;
|
||||
ret << header;
|
||||
ret << buildPacket(id);
|
||||
/*for (int i = 0; i < session[id].size_s(); i++) {
|
||||
EntryInfo fi = session[id][i];
|
||||
// piCoutObj << "send" << fi;
|
||||
bytes_total_cur += fi.fsize;
|
||||
ret << fi;
|
||||
if (fi.entry.size > 0) {
|
||||
PIString path = dir.absolutePath() + dir.separator + fi.entry.path;
|
||||
if (work_file.path() != path || !work_file.isOpened()) {
|
||||
if (!work_file.open(path, PIIODevice::ReadOnly)) {
|
||||
break_ = true;
|
||||
state_string = "ERROR! while open file " + fi.entry.path;
|
||||
piCoutObj << state_string;
|
||||
return ret;
|
||||
}
|
||||
}
|
||||
work_file.seek(fi.fstart);
|
||||
ba.resize(fi.fsize);
|
||||
int rs = work_file.read(ba.data(), ba.size());
|
||||
if (rs != fi.fsize) {
|
||||
break_ = true;
|
||||
state_string = "ERROR! while read file " + fi.entry.path + " (must " + PIString::fromNumber(fi.fsize) + ", but read " + PIString::fromNumber(rs) + ")";
|
||||
piCoutObj << state_string;
|
||||
return ret;
|
||||
}
|
||||
ret << ba;
|
||||
}
|
||||
}
|
||||
EntryInfo cfile = session[id].back();
|
||||
state_string = "sending: " + cfile.entry.path;
|
||||
bytes_file_all = cfile.entry.size;
|
||||
bytes_file_cur = cfile.fstart;
|
||||
uint scrc = crc.calculate(ret);
|
||||
ret << scrc;*/
|
||||
//piCoutObj << "packet" << header.id << "send crc" << scrc;
|
||||
return ret;
|
||||
}
|
||||
|
||||
|
||||
bool PIBaseTransfer::finish_send(bool ok) {
|
||||
if (ok) state_string = "send done";
|
||||
else state_string = "send failed";
|
||||
// piCoutObj << state_string << PIString::readableSize(bytes_total_all);
|
||||
is_sending = false;
|
||||
bytes_all = bytes_cur = 0;
|
||||
header.id = 0;
|
||||
if (!ok) sendBreak(header.session_id);
|
||||
else sendReply(pt_ReplySuccess);
|
||||
finishSend(ok);
|
||||
return ok;
|
||||
}
|
||||
|
||||
|
||||
void PIBaseTransfer::finish_receive(bool ok) {
|
||||
if (ok) state_string = "receive done";
|
||||
else state_string = "receive failed";
|
||||
// piCoutObj << state_string << PIString::readableSize(bytes_total_all);
|
||||
is_receiving = false;
|
||||
bytes_all = bytes_cur = 0;
|
||||
if (!ok) sendBreak(header.session_id);
|
||||
finishReceive(ok);
|
||||
}
|
||||
PIByteArray PIDataTransfer::buildPacket(int id) {
|
||||
;
|
||||
}
|
||||
|
||||
|
||||
Reference in New Issue
Block a user