transfers patch

libs/main/io_utils/pibasetransfer.cpp

@@ -234,6 +234,7 @@ void PIBaseTransfer::received(PIByteArray data) {
 				mutex_send.lock();
 				send_queue = 0;
 				mutex_send.unlock();
+				beginReceive();
 				receiveStarted();
 				state_string = "receiving";
 				replies[0] = pt_ReplySuccess;

libs/main/io_utils/pibasetransfer.h

@@ -93,6 +93,7 @@ protected:
 	void buildSession(PIVector<Part> parts);
 	virtual PIByteArray buildPacket(Part fi) = 0;
 	virtual void receivePart(Part fi, PIByteArray ba, PIByteArray pheader) = 0;
+	virtual void beginReceive() {;}
 	virtual PIByteArray customHeader() {return PIByteArray();}
 	bool send_process();
 	

libs/main/io_utils/pidatatransfer.cpp

@@ -34,6 +34,11 @@ void PIDataTransfer::receivePart(Part fi, PIByteArray ba, PIByteArray pheader) {
 }
 
 
+void PIDataTransfer::beginReceive() {
+	data_.clear();
+}
+
+
 bool PIDataTransfer::send(const PIByteArray& ba) {
 	data_ = ba;
 	buildSession(PIVector<Part>() << Part(0, data_.size()));

libs/main/io_utils/pidatatransfer.h

@@ -39,6 +39,7 @@ public:
 private:
 	virtual PIByteArray buildPacket(Part p);
 	virtual void receivePart(PIBaseTransfer::Part fi, PIByteArray ba, PIByteArray pheader);
+	virtual void beginReceive();
 	PIByteArray data_;
 };
 

libs/main/io_utils/pifiletransfer.cpp

@@ -30,7 +30,6 @@ PIFileTransfer::PIFileTransfer() {
 	dir = PIDir::current();
 	started_ = scanning = false;
 	bytes_file_all = bytes_file_cur = 0;
-	CONNECT(void, this, receiveStarted, this, receive_started);
 	CONNECT1(void, bool, this, sendFinished, this, send_finished);
 	CONNECT1(void, bool, this, receiveFinished, this, receive_finished);
 }
@@ -286,7 +285,7 @@ PIByteArray PIFileTransfer::customHeader() {
 }
 
 
-void PIFileTransfer::receive_started() {
+void PIFileTransfer::beginReceive() {
 	if (pftheader.step == pft_None) {
 		files_.clear();
 //		piCoutObj << "start receive"

libs/main/io_utils/pifiletransfer.h

@@ -103,7 +103,7 @@ private:
 	virtual void receivePart(Part fi, PIByteArray ba, PIByteArray pheader);
 	virtual PIByteArray buildPacket(Part fi);
 	virtual PIByteArray customHeader();
-	EVENT_HANDLER(void, receive_started);
+	virtual void beginReceive();
 	EVENT_HANDLER1(void, send_finished, bool, ok);
 	EVENT_HANDLER1(void, receive_finished, bool, ok);
 };

main.cpp

@@ -1,9 +1,70 @@
 #include "pip.h"
 
 
-
+PIPeer p0("p0"), p1("p1");
 
 int main() {
+
+	PIMap<uint, int> sends, recs;
+	PISet<uint> errors, missed;
+
+	CONNECTL(&p0, dataReceivedEvent, ([&](const PIString & from, const PIByteArray & data){
+		uint cnt = *(uint*)(data.data());
+		piCout << "rec " << cnt << data.size_s();
+		recs[cnt] = data.size_s();
+		if (sends[cnt] != data.size_s()) {
+			piCout << "    " << cnt << "ERROR";
+			errors << cnt;
+		}
+	}));
+
+	p0.start();
+	p1.start();
+
+	piSleep(2);
+
+	int count = 10;
+	for (int i = 0; i < count; ++i) {
+		PIByteArray msg;
+		msg << (uint)i;
+		msg.enlarge(randomi() % 3800 + 4000);
+		sends[i] = msg.size_s();
+		piCout << "send" << i << msg.size_s();
+		p1.send(p0.name(), msg);
+		piMSleep(100);
+	}
+
+	piSleep(1);
+	for (int i = 0; i < count; ++i)
+		if (!recs.contains(i)) missed << i;
+	piCout << "errors" << errors;
+	piCout << "missed" << missed;
+
+
+	/*PIDataTransfer tr0, tr1;
+	CONNECTL(&tr0, receiveFinished, ([&](bool ok){
+		PIByteArray ba = tr0.data();
+		uint cnt = *(uint*)(ba.data());
+		piCout << "rec " << cnt << ok << ba.size_s();
+	}));
+
+	CONNECTL(&tr1, sendRequest, ([&](PIByteArray & data){tr0.received(data);}));
+	CONNECTL(&tr0, sendRequest, ([&](PIByteArray & data){tr1.received(data);}));
+
+	int count = 10;
+	for (int i = 0; i < count; ++i) {
+		PIByteArray msg;
+		msg << (uint)i;
+		msg.enlarge(randomi() % 3800 + 4000);
+		//sends[i] = msg.size_s();
+		piCout << "send" << i << msg.size_s();
+		tr1.send(msg);
+		//piMSleep(100);
+	}*/
+
+
+	return 0;
+
 	const int iters = 4;
 	const int sz = 10000000;