pip/src/io/pibasetransfer.cpp

#include "pibasetransfer.h"

const uint PIBaseTransfer::signature = 0x54424950;

PIBaseTransfer::PIBaseTransfer(): crc(standardCRC_16()), diag(false) {
	header.sig = signature;
	header.session_id = 0;
	packet_header_size = sizeof(PacketHeader) + customHeader().size();
	part_header_size = sizeof(Part) + sizeof(int);
	is_sending = is_receiving = is_pause = false;
	break_ = true;
	bytes_all = bytes_cur = 0;
	replies_cnt = send_queue = 0;
	timeout_ = 10.;
	diag.setDisconnectTimeout(timeout_);
	CONNECTU(&diag, qualityChanged, this, diagChanged);
	packets_count = 32;
	setPacketSize(4096);
	randomize();
}


PIBaseTransfer::~PIBaseTransfer() {
	diag.stop();
	break_ = true;
}


void PIBaseTransfer::stopSend() {
	if (!is_sending) return;
	break_ = true;
}


void PIBaseTransfer::stopReceive() {
	if (!is_receiving) return;
	break_ = true;
	//piCoutObj << "stopReceive()";
	finish_receive(false);
}


void PIBaseTransfer::setPause(bool pause_) {
	if (is_pause == pause_) return;
	pause_tm.reset();
	is_pause = pause_;
	if (pause_) paused();
	else resumed();
}


void PIBaseTransfer::received(PIByteArray data) {
	packet_header_size = sizeof(PacketHeader) + customHeader().size();
	if (data.size() < sizeof(PacketHeader)) {
		diag.received(data.size(), false);
		return;
	}
	PacketHeader h;
	data >> h;
	PacketType pt = (PacketType)h.type;
	if (!h.check_sig()) {
		diag.received(data.size(), false);
		return;
	} else diag.received(data.size(), true);
	//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(), data.size_s());
				if (rcrc != ccrc) {
					header.id = h.id;
					sendReply(pt_ReplyInvalid);
					piCoutObj << "invalid CRC";
				} else {
					processData(h.id, data);
				}
				if (is_pause) sendReply(pt_Pause);
			}
		break;
		case pt_ReplySuccess:
		case pt_ReplyInvalid:
			if (h.session_id != header.session_id) return;
			if (is_pause) sendReply(pt_Pause);
			if (pt == pt_ReplySuccess) {
				send_queue--;
				if (send_queue < 0) send_queue = 0;
				send_tm.reset();
			}
			replies_cnt++;
			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);
				//if (checkSession() == 0 && pt == pt_ReplySuccess) { piCoutObj << "Success receive"; finish_receive(true);}
			}
		break;
		case pt_Break:
			break_ = true;
			// piCoutObj << "BREAK";
			if (is_receiving) {
				stopReceive();
				return;
			}
			if (is_sending) {
				stopSend();
				return;
			}
		break;
		case pt_Start:
			if (is_pause && (is_sending || is_receiving)) {
				if (header.session_id == h.session_id) {
					is_pause = false;
					resumed();
					return;
				}
			}
			if (is_sending && header.session_id != h.session_id) {
				sendBreak(h.session_id);
				return;
			}
			if (is_receiving) {
				if (header.session_id != h.session_id) {
					//sendBreak(h.session_id);
					//return;
					//piCoutObj << "restart receive";
					finish_receive(false, true);
				} else return;
			}
			if (data.size() == sizeof(StartRequest)) {
				StartRequest sr;
				data >> sr;
				bytes_cur = 0;
				state_string = "start request";
				bytes_all = sr.size;
				header.session_id = h.session_id;
				header.id = 0;
				session.clear();
				replies.clear();
				session.resize(sr.packets);
				replies.resize(sr.packets + 1);
				replies.fill(pt_Unknown);
				diag.reset();
				diag.start(100);
				//piCoutObj << "receiveStarted()";
				is_receiving = true;
				break_ = false;
				receiveStarted();
				replies_cnt = send_queue = 0;
				state_string = "receiving";
				replies[0] = pt_ReplySuccess;
				sendReply(pt_ReplySuccess);
			}
		break;
		case pt_Pause:
			if (header.session_id == h.session_id) {
// 				piCout << "receive pause";
				if (!is_pause && pause_tm.elapsed_s() < timeout_/10) {
// 					piCout << "resume";
					sendReply(pt_Start);
					return;
				}
				if (!is_pause) paused();
				is_pause = true;
				if (is_receiving && pause_tm.elapsed_m() > 40) {
// 					piCout << "send pause";
					sendReply(pt_Pause);
				}
				if (is_sending) send_tm.reset();
				pause_tm.reset();
			}
			break;
		default: break;
	}
}


bool PIBaseTransfer::send_process() {
	packet_header_size = sizeof(PacketHeader) + customHeader().size();
	break_ = false;
	diag.reset();
	diag.start(100);
	sendStarted();
	is_sending = true;
	replies.resize(session.size() + 1);
	replies.fill(pt_Unknown);
	PIByteArray ba;
	if (!getStartRequest()) return finish_send(false);
	replies_cnt = send_queue = 0;
	state_string = "sending";
	PITimeMeasurer stm;
	send_tm.reset();
	//int ltm = 0;
	for (int i = 0; i < session.size_s(); i++) {
		if (send_queue >= packets_count || is_pause) {
			--i;
			piMSleep(1);
			if (is_pause) {
				piMSleep(40);
// 				piCout << "send pause";
				sendReply(pt_Pause);
				if (pause_tm.elapsed_s() > timeout())return finish_send(false);
			}
			if (send_tm.elapsed_s() > timeout_) return finish_send(false);
			if (stm.elapsed_s() > timeout_ / 10.)
				send_queue = 0;
			continue;
		}
		stm.reset();
		ba = build_packet(i);
		diag.sended(ba.size());
		sendRequest(ba);
		send_queue++;
		if (break_) return finish_send(false);
	}
	PITimeMeasurer tm2, tm;
	int prev_chk = 0;
	send_queue = 0;
	while (tm.elapsed_s() < timeout_) {
		int chk = checkSession();
		if (chk == 0) return finish_send(true);
		if (chk != prev_chk) tm.reset();
		else if (tm.elapsed_m() < 100) {
			piMSleep(1);
			continue;
		}
		if (is_pause) {
			piMSleep(40);
// 			piCout << "send pause";
			sendReply(pt_Pause);
			if (pause_tm.elapsed_s() > timeout())return finish_send(false);
			else continue;
		}
		prev_chk = chk;
		if (chk > 0) {
			if (tm2.elapsed_s() > timeout_ / 10.) {
				//piCoutObj << "recovery packet" << chk;
				if (send_queue >= packets_count) {
					piMSleep(10);
					continue;
				}
				ba = build_packet(chk - 1);
				diag.sended(ba.size());
				sendRequest(ba);
				send_queue++;
			}
		}
		//		if (chk == -1) return finish_send(false);
		if (break_) return finish_send(false);
		piMSleep(1);
	}
	return finish_send(false);
}


int PIBaseTransfer::checkSession() {
	if (!(is_receiving || is_sending)) return -1;
	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 PIBaseTransfer::buildSession(PIVector<Part> parts) {
	state_string = "calculating parts ... ";
	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 parts ... " + PIString::fromNumber(i) + " of " + PIString::fromNumber(parts.size());
		fi.id = parts[i].id;
		//piCout << fi.id << state_string;
		bytes_all += parts[i].size;
		//fi.size = fi.entry.size;
		fi.start = 0;
		llong rest = parts[i].size - (packet_size - cur_size);
		//piCout << i << fi.size << rest << bytes_all;
		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;
}


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;
	if (is_sending || is_receiving) diag.sended(ba.size());
	sendRequest(ba);
}


bool PIBaseTransfer::getStartRequest() {
	PITimeMeasurer tm;
	header.type = pt_Start;
	header.id = 0;
	PIByteArray ba;
	StartRequest st;
	st.packets = (uint)session.size();
	st.size = bytes_all;
	ba << header;
	ba << st;
	state_string = "send request";
	while (tm.elapsed_s() < timeout_) {
		diag.sended(ba.size());
		sendRequest(ba);
		if (break_) return false;
		//piCoutObj << send_replyes[0];
		if (replies[0] == pt_ReplySuccess) {
			state_string = "send permited!";
			return true;
		}
		piMSleep(100);
	}
	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 << "session id" << header.session_id;
	//ret << header;
	ret.append(customHeader());
	for (int i = 0; i < session[id].size_s(); i++) {
		Part fi = session[id][i];
		ret << fi;
		//piCout << "biuld" << fi;
		ba = buildPacket(fi);
		bytes_cur += ba.size();
		if (ba.size() != fi.size) piCoutObj << "***error while build packet, wrong part size";
		ret << ba;
	}
	header.crc = crc.calculate(ret);
	PIByteArray hdr; hdr << header;
	ret.insert(0, hdr);
	//piCoutObj << "Packet" << header.id << ret.size();
	return ret;
}


bool PIBaseTransfer::finish_send(bool ok) {
	is_sending = false;
	if (ok) state_string = "send done";
	else state_string = "send failed";
	//piCoutObj << state_string << PIString::readableSize(bytes_all);
	header.id = 0;
	if (!ok) sendBreak(header.session_id);
	else sendReply(pt_ReplySuccess);
	sendFinished(ok);
	diag.stop();
	bytes_all = bytes_cur = 0;
	return ok;
}


void PIBaseTransfer::finish_receive(bool ok, bool quet) {
	is_receiving = false;
	if (ok) state_string = "receive done";
	else state_string = "receive failed";
	//piCoutObj << state_string;// << PIString::readableSize(bytes_all);
	if (!ok && !quet) sendBreak(header.session_id);
	receiveFinished(ok);
	diag.stop();
	bytes_all = bytes_cur = 0;
}


void PIBaseTransfer::diagChanged(PIDiagnostics::Quality new_quality, PIDiagnostics::Quality old_quality) {
	if (is_receiving) {
		if (new_quality == PIDiagnostics::Failure) {
			piCout << "disconnected!";
			break_ = true;
			is_receiving = false;
			state_string = "receive failed";
			receiveFinished(false);
			bytes_all = bytes_cur = 0;
			//stopReceive();
		}
	}
}