2 files changed, 81 insertions, 52 deletions

diff --git a/qacp.c b/qacp.c
index 3afc35e..295c2d9 100644
--- a/qacp.c
+++ b/qacp.c
@@ -148,6 +148,11 @@ qacp_sendonefile(const char *host, const char *srcfile, const char *destfile)
 	struct stat sb;
 	const int bufsize = QARSH_MAX_PACKET_SIZE/2;
 	char buf[bufsize];
+	fd_set rfds, wfds;
+	int nfd;
+	struct timeval timeout;
+	int dallow;
+	int complete;
 
 
 	if ((fd = open(srcfile, O_RDONLY)) <0) {
@@ -174,59 +179,80 @@ qacp_sendonefile(const char *host, const char *srcfile, const char *destfile)
 	send_packet(qacp_fd, qp);
 	qpfree(qp);
 
-	/* Await our return code from qarshd */
-	qp = recv_packet(qacp_fd);
-	if (qp && qp->qp_type == QP_RETURNCODE) {
-		if (qp->qp_returncode.qp_rc != 0) {
-			fprintf(stderr, "Remote side failed: %s\n",
-					qp->qp_returncode.qp_strerror);
-			qpfree(qp);
-			goto sendone_error;
-		}
-		qpfree(qp);
-	} else {
-		fprintf(stderr, "Did not receive response to recvfile\n");
-		goto sendone_failure;
-	}
-
 	offset = 0;
-	do {
-		nbytes = read(fd, buf, bufsize);
-		if (nbytes < 0) {
-			fprintf(stderr, "read() error: %s\n", strerror(errno));
-			qp = make_qp_returncode(-1, errno, strerror(errno));
-		} else {
-			qp = make_qp_data(0, offset, nbytes, buf);
+	dallow = 0;
+	complete = 0;
+	for (;;) {
+		FD_ZERO(&rfds);
+		FD_SET(qacp_fd, &rfds);
+		FD_ZERO(&wfds);
+		if (!complete && dallow > 0) FD_SET(qacp_fd, &wfds);
+		timeout.tv_sec = 10;
+		timeout.tv_usec = 0;
+
+		nfd = select(qacp_fd+1, &rfds, &wfds, NULL, &timeout);
+		if (nfd < 0) {
+			if (errno == EINTR) {
+				continue;
+			} else {
+				fprintf(stderr, "select errno %d, %s\n", errno, strerror(errno));
+			}
+		} else if (nfd > 0) {
+			if (nfd && FD_ISSET(qacp_fd, &rfds)) {
+				qp = recv_packet(qacp_fd);
+				if (qp == NULL) {
+					fprintf(stderr, "No packet\n");
+					break;
+				}
+				if (qp->qp_type == QP_DALLOW) {
+					if (qp->qp_dallow.qp_count > 0) {
+						dallow += qp->qp_dallow.qp_count;
+					} else {
+						dallow = 0;
+					}
+				} else if (qp->qp_type == QP_RETURNCODE) {
+					if (qp->qp_returncode.qp_rc == 0) { /* success */
+						break;
+					} else if (qp->qp_returncode.qp_rc == -1) {
+						fprintf(stderr, "Transfer failed: %s\n", 
+								qp->qp_returncode.qp_strerror);
+						exit(125);
+					}
+				} else {
+					fprintf(stderr, "Received unexpected packet\n");
+					dump_qp(qp);
+					break;
+				}
+				qpfree(qp);
+			}
+			if (nfd && FD_ISSET(qacp_fd, &wfds)) {
+				/* Create a data packet to send */
+				nbytes = read(fd, buf, dallow);
+				if (nbytes < 0) {
+					fprintf(stderr, "read() error: %s\n",
+							strerror(errno));
+					qp = make_qp_returncode(-1, errno, strerror(errno));
+				} else {
+					qp = make_qp_data(0, offset, nbytes, buf);
+					if (nbytes == 0) { /* Transfer is complete */
+						complete = 1;
+					} else {
+						offset += nbytes;
+						dallow -= nbytes;
+					}
+				}
+				send_packet(qacp_fd, qp);
+				qpfree(qp);
+			}
 		}
-		send_packet(qacp_fd, qp);
-		offset += nbytes;
-		qpfree(qp);
-	} while (nbytes > 0);
-sendone_failure:
-
-	close(fd);
-
-	/* Await our return code from qarshd */
-	qp = recv_packet(qacp_fd);
-	if (qp && qp->qp_type == QP_RETURNCODE 
-			&& qp->qp_returncode.qp_rc == -1) {
-		fprintf(stderr, "Remote side failed, %s\n",
-				qp->qp_returncode.qp_strerror);
-		close(qacp_fd);
-		qpfree(qp);
-		exit(125);
 	}
-
-	qpfree(qp);
+	close(fd);
 
 	if (!quiet) {
 		printf("%-30.30s  -> %s:%s\n", srcfile, host, destfile);
 	}
 
 	return;
-sendone_error:
-	close(qacp_fd);
-	exit(125);
 }
 
 
diff --git a/qarshd.c b/qarshd.c
index d686078..78b9dda 100644
--- a/qarshd.c
+++ b/qarshd.c
@@ -183,13 +183,13 @@ prepare_recvfile(struct qa_packet *qp)
 	recvsize = qp->qp_recvfile.qp_count;
 	received = 0;
 
-	return make_qp_returncode(0, 0, "Ready to receive");
+	return make_qp_data_allow(0, QARSHD_BUFSIZE);
 }
 
 struct qa_packet *
 receive_data(struct qa_packet *qp)
 {
-	ssize_t nwrote;
+	ssize_t ret, nwrote;
 
 	if (qp->qp_data.qp_count == 0) { /* EOF */
 		close(receivefd);
@@ -198,13 +198,16 @@ receive_data(struct qa_packet *qp)
 	}
 	if (debug) syslog(LOG_DEBUG, "Data for %d, %d@%ld\n",
 			receivefd, qp->qp_data.qp_count, qp->qp_data.qp_offset);
-	nwrote = write(receivefd, qp->qp_data.qp_blob, qp->qp_data.qp_count);
-
-	if (nwrote < 0) {
-		return make_qp_returncode(-1, errno, strerror(errno));
-	}
+	nwrote = 0;
+	do {
+		ret = write(receivefd, qp->qp_data.qp_blob + nwrote, qp->qp_data.qp_count - nwrote);
+		if (ret < 0) {
+			return make_qp_returncode(-1, errno, strerror(errno));
+		} 
+		nwrote += ret;
+	} while (nwrote < qp->qp_data.qp_count);
 	received += nwrote;
-	return NULL;
+	return make_qp_data_allow(0, nwrote);
 }
 
 ssize_t