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

git-svn-id: svn://db.shs.com.ru/pip@104 12ceb7fc-bf1f-11e4-8940-5bc7170c53b5

Browse Source
This commit is contained in:
Бычков Андрей
2015-04-15 14:29:54 +00:00
parent 368325e53c
commit 022b76bc29
1 changed files with 25 additions and 23 deletions
Download Patch File Download Diff File Expand all files Collapse all files

48
src/io/pibasetransfer.cpp
View File

@@ -35,7 +35,7 @@ void PIBaseTransfer::stopSend() {
void PIBaseTransfer::stopReceive() { void PIBaseTransfer::stopReceive() {
if (!is_receiving) return; if (!is_receiving) return;
break_ = true; break_ = true;
//piCoutObj << "stopReceive()"; piCoutObj << "stopReceive()";
finish_receive(false); finish_receive(false);
} }
@@ -62,7 +62,7 @@ void PIBaseTransfer::received(PIByteArray data) {
diag.received(data.size(), false); diag.received(data.size(), false);
return; return;
} else diag.received(data.size(), true); } else diag.received(data.size(), true);
//piCoutObj << "receive" << h.session_id << h.type << h.id; piCoutObj << "receive" << h.session_id << h.type << h.id;
switch (pt) { switch (pt) {
case pt_Unknown: break; case pt_Unknown: break;
case pt_Data: case pt_Data:
@@ -97,13 +97,13 @@ void PIBaseTransfer::received(PIByteArray data) {
replies[h.id] = pt; replies[h.id] = pt;
} }
if (is_receiving && h.id == 0) { if (is_receiving && h.id == 0) {
if (checkSession() == 0 && pt == pt_ReplySuccess) finish_receive(true); // if (checkSession() == 0 && pt == pt_ReplySuccess) finish_receive(true);
//if (checkSession() == 0 && pt == pt_ReplySuccess) { piCoutObj << "Success receive"; finish_receive(true);} if (checkSession() == 0 && pt == pt_ReplySuccess) { piCoutObj << "Success receive"; finish_receive(true);}
} }
break; break;
case pt_Break: case pt_Break:
break_ = true; break_ = true;
// piCoutObj << "BREAK"; piCoutObj << "BREAK";
if (is_receiving) { if (is_receiving) {
stopReceive(); stopReceive();
return; return;
@@ -129,7 +129,7 @@ void PIBaseTransfer::received(PIByteArray data) {
if (header.session_id != h.session_id) { if (header.session_id != h.session_id) {
//sendBreak(h.session_id); //sendBreak(h.session_id);
//return; //return;
//piCoutObj << "restart receive"; piCoutObj << "restart receive";
finish_receive(false, true); finish_receive(false, true);
} else return; } else return;
} }
@@ -148,7 +148,7 @@ void PIBaseTransfer::received(PIByteArray data) {
replies.fill(pt_Unknown); replies.fill(pt_Unknown);
diag.reset(); diag.reset();
diag.start(100); diag.start(100);
//piCoutObj << "receiveStarted()"; piCoutObj << "receiveStarted()";
is_receiving = true; is_receiving = true;
break_ = false; break_ = false;
receiveStarted(); receiveStarted();
@@ -160,16 +160,16 @@ void PIBaseTransfer::received(PIByteArray data) {
break; break;
case pt_Pause: case pt_Pause:
if (header.session_id == h.session_id) { if (header.session_id == h.session_id) {
// piCout << "receive pause"; piCout << "receive pause";
if (!is_pause && pause_tm.elapsed_s() < timeout_/10) { if (!is_pause && pause_tm.elapsed_s() < timeout_/10) {
// piCout << "resume"; piCout << "resume";
sendReply(pt_Start); sendReply(pt_Start);
return; return;
} }
if (!is_pause) paused(); if (!is_pause) paused();
is_pause = true; is_pause = true;
if (is_receiving && pause_tm.elapsed_m() > 40) { if (is_receiving && pause_tm.elapsed_m() > 40) {
// piCout << "send pause"; piCout << "send pause";
sendReply(pt_Pause); sendReply(pt_Pause);
} }
if (is_sending) send_tm.reset(); if (is_sending) send_tm.reset();
@@ -203,7 +203,7 @@ bool PIBaseTransfer::send_process() {
piMSleep(1); piMSleep(1);
if (is_pause) { if (is_pause) {
piMSleep(40); piMSleep(40);
// piCout << "send pause"; piCout << "send pause";
sendReply(pt_Pause); sendReply(pt_Pause);
if (pause_tm.elapsed_s() > timeout())return finish_send(false); if (pause_tm.elapsed_s() > timeout())return finish_send(false);
} }
@@ -232,7 +232,7 @@ bool PIBaseTransfer::send_process() {
} }
if (is_pause) { if (is_pause) {
piMSleep(40); piMSleep(40);
// piCout << "send pause"; piCout << "send pause";
sendReply(pt_Pause); sendReply(pt_Pause);
if (pause_tm.elapsed_s() > timeout())return finish_send(false); if (pause_tm.elapsed_s() > timeout())return finish_send(false);
else continue; else continue;
@@ -240,7 +240,7 @@ bool PIBaseTransfer::send_process() {
prev_chk = chk; prev_chk = chk;
if (chk > 0) { if (chk > 0) {
if (tm2.elapsed_s() > timeout_ / 10.) { if (tm2.elapsed_s() > timeout_ / 10.) {
//piCoutObj << "recovery packet" << chk; piCoutObj << "recovery packet" << chk;
if (send_queue >= packets_count) { if (send_queue >= packets_count) {
piMSleep(10); piMSleep(10);
continue; continue;
@@ -289,12 +289,12 @@ void PIBaseTransfer::buildSession(PIVector<Part> parts) {
for (int i = 0; i < parts.size_s(); i++) { for (int i = 0; i < parts.size_s(); i++) {
state_string = "calculating parts ... " + PIString::fromNumber(i) + " of " + PIString::fromNumber(parts.size()); state_string = "calculating parts ... " + PIString::fromNumber(i) + " of " + PIString::fromNumber(parts.size());
fi.id = parts[i].id; fi.id = parts[i].id;
//piCout << fi.id << state_string; piCout << fi.id << state_string;
bytes_all += parts[i].size; bytes_all += parts[i].size;
//fi.size = fi.entry.size; //fi.size = fi.entry.size;
fi.start = 0; fi.start = 0;
llong rest = parts[i].size - (packet_size - cur_size); llong rest = parts[i].size - (packet_size - cur_size);
//piCout << i << fi.size << rest << bytes_all; piCout << i << fi.size << rest << bytes_all;
if (rest <= 0) { if (rest <= 0) {
fi.size = parts[i].size; fi.size = parts[i].size;
lfi << fi; lfi << fi;
@@ -334,6 +334,7 @@ void PIBaseTransfer::buildSession(PIVector<Part> parts) {
void PIBaseTransfer::sendBreak(int session_id) { void PIBaseTransfer::sendBreak(int session_id) {
piCoutObj << "sendBreak";
uint psid = header.session_id; uint psid = header.session_id;
header.session_id = session_id; header.session_id = session_id;
sendReply(pt_Break); sendReply(pt_Break);
@@ -342,6 +343,7 @@ void PIBaseTransfer::sendBreak(int session_id) {
void PIBaseTransfer::sendReply(PacketType reply) { void PIBaseTransfer::sendReply(PacketType reply) {
piCoutObj << "sendReply" << reply;
header.type = reply; header.type = reply;
PIByteArray ba; PIByteArray ba;
ba << header; ba << header;
@@ -365,7 +367,7 @@ bool PIBaseTransfer::getStartRequest() {
diag.sended(ba.size()); diag.sended(ba.size());
sendRequest(ba); sendRequest(ba);
if (break_) return false; if (break_) return false;
//piCoutObj << send_replyes[0]; // piCoutObj << replies[0];
if (replies[0] == pt_ReplySuccess) { if (replies[0] == pt_ReplySuccess) {
state_string = "send permited!"; state_string = "send permited!";
return true; return true;
@@ -377,7 +379,7 @@ bool PIBaseTransfer::getStartRequest() {
void PIBaseTransfer::processData(int id, PIByteArray & data) { void PIBaseTransfer::processData(int id, PIByteArray & data) {
//piCoutObj << "received packet" << id << ", size" << data.size(); piCoutObj << "received packet" << id << ", size" << data.size();
if (id < 1 || id > replies.size_s()) return; if (id < 1 || id > replies.size_s()) return;
if (!session[id - 1].isEmpty()) { if (!session[id - 1].isEmpty()) {
header.id = id; header.id = id;
@@ -400,7 +402,7 @@ void PIBaseTransfer::processData(int id, PIByteArray & data) {
data >> ba; data >> ba;
//fi.fsize = ba.size(); //fi.fsize = ba.size();
bytes_cur += fi.size; bytes_cur += fi.size;
//piCoutObj << "recv" << fi; piCoutObj << "recv" << fi;
session[id - 1] << fi; session[id - 1] << fi;
state_string = "receiving..."; state_string = "receiving...";
receivePart(fi, ba, pheader); receivePart(fi, ba, pheader);
@@ -417,13 +419,13 @@ PIByteArray PIBaseTransfer::build_packet(int id) {
PIByteArray ba; PIByteArray ba;
header.id = id + 1; header.id = id + 1;
header.type = pt_Data; header.type = pt_Data;
//piCoutObj << "session id" << header.session_id; piCoutObj << "session id" << header.session_id;
//ret << header; //ret << header;
ret.append(customHeader()); ret.append(customHeader());
for (int i = 0; i < session[id].size_s(); i++) { for (int i = 0; i < session[id].size_s(); i++) {
Part fi = session[id][i]; Part fi = session[id][i];
ret << fi; ret << fi;
//piCout << "biuld" << fi; piCout << "biuld" << fi;
ba = buildPacket(fi); ba = buildPacket(fi);
bytes_cur += ba.size(); bytes_cur += ba.size();
if (ba.size() != fi.size) piCoutObj << "***error while build packet, wrong part size"; if (ba.size() != fi.size) piCoutObj << "***error while build packet, wrong part size";
@@ -432,7 +434,7 @@ PIByteArray PIBaseTransfer::build_packet(int id) {
header.crc = crc.calculate(ret); header.crc = crc.calculate(ret);
PIByteArray hdr; hdr << header; PIByteArray hdr; hdr << header;
ret.insert(0, hdr); ret.insert(0, hdr);
//piCoutObj << "Packet" << header.id << ret.size(); piCoutObj << "Packet" << header.id << ret.size();
return ret; return ret;
} }
@@ -441,7 +443,7 @@ bool PIBaseTransfer::finish_send(bool ok) {
is_sending = false; is_sending = false;
if (ok) state_string = "send done"; if (ok) state_string = "send done";
else state_string = "send failed"; else state_string = "send failed";
//piCoutObj << state_string << PIString::readableSize(bytes_all); piCoutObj << state_string << PIString::readableSize(bytes_all);
header.id = 0; header.id = 0;
if (!ok) sendBreak(header.session_id); if (!ok) sendBreak(header.session_id);
else sendReply(pt_ReplySuccess); else sendReply(pt_ReplySuccess);
@@ -456,7 +458,7 @@ void PIBaseTransfer::finish_receive(bool ok, bool quet) {
is_receiving = false; is_receiving = false;
if (ok) state_string = "receive done"; if (ok) state_string = "receive done";
else state_string = "receive failed"; else state_string = "receive failed";
//piCoutObj << state_string;// << PIString::readableSize(bytes_all); piCoutObj << state_string << PIString::readableSize(bytes_all);
if (!ok && !quet) sendBreak(header.session_id); if (!ok && !quet) sendBreak(header.session_id);
receiveFinished(ok); receiveFinished(ok);
diag.stop(); diag.stop();