Get qacp local to remote working on main socket

I removed the buffering layer from recv_packet because
it made the logic too complex around the pselect in qarshd.
Now only read as much as needed to get each packet.

qarshd adds an array for remote file descriptors which is
only a stub for now.  This needs to be expanded to allow
multiple file transfers at the same time for runcmd.

1 files changed, 65 insertions, 72 deletions

diff --git a/qarshd.c b/qarshd.c
index 440c3ec..a81055b 100644
--- a/qarshd.c
+++ b/qarshd.c
@@ -47,7 +47,6 @@
 
 int debug = 0;
 
-
 /* Globals */
 struct sockaddr_storage peername;
 int child_exitted = 0;
@@ -57,6 +56,12 @@ sigset_t orig_sigmask;
 char *saved_path = NULL;
 struct stat saved_stat;
 
+struct remotefd {
+	int fd;
+	int direction; /* 0 incoming, 1 outgoing */
+	off_t expectedsize;
+	off_t received;
+} remotefds[8];
 
 int
 setup_user(char *user, char *group)
@@ -126,58 +131,53 @@ run_cmd(const char *cmd, int p_in, int p_out, int p_err)
 	return pid;
 }
 
-off_t 
-recvfile(const char *path, int if_port, off_t count, mode_t mode)
+struct qa_packet *
+prepare_recvfile(struct qa_packet *qp)
 {
-	int sd;
-	int ofd;
-	char buf[BUFSIZ];
-	ssize_t nread, nwrote;
-	off_t nleft;
-
-	/* Read count bytes from ifd (sd after we connect), 
-	 * write into file @ path 
-	 */
-
-	sd = connect_to_peer(&peername, if_port);
-	if (sd == -1) {
-		syslog(LOG_WARNING, "connect to if_port failed\n");
-		return -1;
+	int fd;
+
+	syslog(LOG_INFO, "Receive file %s, size = %lld, mode = %o\n",
+		qp->qp_recvfile.qp_path,
+		(long long int)qp->qp_recvfile.qp_count,
+		qp->qp_recvfile.qp_mode);
+
+	if ((fd = open(qp->qp_recvfile.qp_path, O_TRUNC|O_CREAT|O_WRONLY,
+					qp->qp_recvfile.qp_mode)) < 0) {
+		syslog(LOG_WARNING, "Could not open %s to receive file: %s\n",
+				qp->qp_recvfile.qp_path, strerror(errno));
+		return make_qp_returncode(-1, errno, strerror(errno));
 	}
 
-	if ((ofd = open(path, O_TRUNC|O_CREAT|O_WRONLY, mode)) < 0) {
-		syslog(LOG_WARNING, "Could not open %s to recv file: %s\n",
-		       path, strerror(errno));
-		return -1;
-	}
-	
-	fchmod(ofd, mode);
-
-	nleft = count;
-	while (nleft > 0) {
-		nread = read(sd, buf, BUFSIZ);
-		if (nread < 0) {
-			return nread;
-		} else if (nread == 0) {  /* EOF */
-			break;
-		}
+	/* Set permissions again to override umask */
+	fchmod(fd, qp->qp_recvfile.qp_mode);
 
-		nwrote = write(ofd, buf, nread);
-		nleft -= nread;
-	}
+	/* Store fd to check data packets against */
+	remotefds[0].fd = fd;
+	remotefds[0].direction = 0;
+	remotefds[0].expectedsize = qp->qp_recvfile.qp_count;
+	remotefds[0].received = 0;
 
-	if (nleft != 0) {
-		unlink(path);
-		syslog(LOG_WARNING, "Short file transfer in recvfile(), "
-		       "%lld bytes lost, wanted %lld\n", 
-		       (long long int)nleft, 
-		       (long long int)count);
-	}
+	return make_qp_returncode(0, fd, "Ready to receive");
+}
 
-	close(sd);
-	close(ofd);
+struct qa_packet *
+receive_data(struct remotefd *rfd, struct qa_packet *qp)
+{
+	ssize_t nwrote;
+
+	if (qp->qp_data.qp_count == 0) { /* EOF */
+		close(rfd->fd);
+		syslog(LOG_DEBUG, "Transfer complete\n");
+		return make_qp_returncode(0, 0, "Transfer complete");
+	}
+	/* syslog(LOG_DEBUG, "Data for %d, %d@%ld\n", rfd->fd, qp->qp_data.qp_count, qp->qp_data.qp_offset); */
+	nwrote = write(rfd->fd, qp->qp_data.qp_blob, qp->qp_data.qp_count);
 
-	return count - nleft;
+	if (nwrote < 0) {
+		return make_qp_returncode(-1, errno, strerror(errno));
+	}
+	rfd->received += nwrote;
+	return NULL;
 }
 
 ssize_t
@@ -267,6 +267,10 @@ handle_packets(int infd)
 	sa.sa_mask = sigmask;
 	sa.sa_flags = SA_RESTART;
 	sigaction(SIGCHLD, &sa, NULL);
+#if 0
+	signal(SIGALRM, sig_handler);
+	pause();
+#endif
 
 	for (;;) {
 		FD_SET(infd, &rfds);
@@ -298,7 +302,7 @@ handle_packets(int infd)
 				if (debug) syslog(LOG_DEBUG, "That's enough\n");
 				break;
 			}
-			dump_qp(qp);
+			if (debug) dump_qp(qp);
 			switch (qp->qp_type) {
 			case QP_KILL:
 				if (child_pid) {
@@ -324,38 +328,26 @@ handle_packets(int infd)
 						qp->qp_runcmd.qp_stdout_port,
 						qp->qp_runcmd.qp_stderr_port);
 				break;
-			case QP_RECVFILE: 
-				syslog(LOG_INFO, "Got a QP_RECVFILE with path = %s, "
-				       "ifd = %d, count = %lld, mode = %o\n",
-				        qp->qp_recvfile.qp_path,
-				        qp->qp_recvfile.qp_if_port,
-				        (long long int)qp->qp_recvfile.qp_count,
-				        qp->qp_recvfile.qp_mode);
-				nbytes = recvfile(qp->qp_recvfile.qp_path, 
-				                  qp->qp_recvfile.qp_if_port, 
-				                  qp->qp_recvfile.qp_count, 
- 				                  qp->qp_recvfile.qp_mode);
-				if (nbytes < 0) {
-					rp = make_qp_returncode(-1, errno, strerror(errno));
-				} else if (nbytes < qp->qp_recvfile.qp_count) {
-					char tmpstr[512];
-					sprintf(tmpstr, "Excpected %lld, wrote %lld\n", 
-					        (long long int)qp->qp_recvfile.qp_count, 
-						(long long int)nbytes);
-					rp = make_qp_returncode(-1, 0, tmpstr);
-				} else {
-					rp = make_qp_returncode(0, 0, "Transfer Complete");
-				}
+			case QP_RECVFILE: /* Setup file descriptors to handle incoming data */
+				rp = prepare_recvfile(qp);
 				send_packet(fileno(stdout), rp);
 				qpfree(rp);
 				break;
+			case QP_DATA:
+				assert(qp->qp_data.qp_remfd == remotefds[0].fd);
+				rp = receive_data(&remotefds[0], qp);
+				if (rp) {
+					send_packet(fileno(stdout), rp);
+					qpfree(rp);
+				}
+				break;
 			case QP_SENDFILE:
 				syslog(LOG_INFO, "Got a QP_SENDFILE with path = %s, "
-				       "ofd = %d\n",
+				       "remfd = %d\n",
 				        qp->qp_sendfile.qp_path,
-				        qp->qp_sendfile.qp_of_port);
+				        qp->qp_sendfile.qp_remfd);
 				nbytes = pushfile(qp->qp_sendfile.qp_path, 
-				                  qp->qp_sendfile.qp_of_port);
+				                  qp->qp_sendfile.qp_remfd);
 				if (nbytes < 0) {
 					rp = make_qp_returncode(-1, errno, strerror(errno));
 				} else {
@@ -376,6 +368,7 @@ handle_packets(int infd)
 					"Packet type %s unimplemented",
 					qp_packet_type(qp->qp_type));
 			}
+			qpfree(qp);
 		} else {
 			if (debug) syslog(LOG_DEBUG, "Nothing to do\n");
 		}