upload/download from stdin/stdout

hptp-recv/src/main.rs

@@ -5,7 +5,8 @@ extern crate tokio;
 extern crate thiserror;
 
 use hptp::log::Logger;
-// use tokio::net::UdpSocket;
+use tokio::io::{AsyncWriteExt, Stdout};
+use tokio::net::UdpSocket;
 
 #[derive(Error, Debug)]
 enum Error {
@@ -40,15 +41,17 @@ async fn start(log: &mut Logger) -> Result<(), Error> {
         .await
         .map_err(|_| Error::NoPortAvail)?;
     log.bound(sock.local_addr()?.port()).await;
+    download(log, sock, tokio::io::stdout()).await
+}
 
-    let (mut rx, _tx) = sock.split();
+async fn download(log: &mut Logger, mut sock: UdpSocket, mut out: Stdout) -> Result<(), Error> {
     let mut buf = [0u8; 2000];
+    let mut pos = 0;
     loop {
-        let (n, _who) = rx.recv_from(&mut buf).await?;
-        log.debug_msg(match String::from_utf8(buf[..n].into()) {
-            Ok(s) => format!("got: {:?}", s),
-            Err(e) => format!("got: {:?} (invalid utf8)", e.as_bytes()),
-        })
-        .await;
+        let (len, _who) = sock.recv_from(&mut buf).await?;
+        out.write_all(&buf[..len]).await?;
+        out.flush().await?;
+        log.recv_data_accepted(pos, len, hptp::log::InOrder).await;
+        pos += len;
     }
 }

hptp-send/src/main.rs

@@ -6,6 +6,8 @@ extern crate thiserror;
 
 use hptp::log::Logger;
 use std::net::SocketAddrV4;
+use tokio::io::{AsyncReadExt, Stdin};
+use tokio::net::UdpSocket;
 
 #[derive(Error, Debug)]
 enum Error {
@@ -52,7 +54,7 @@ async fn start(log: &mut Logger) -> Result<(), Error> {
         .map_err(|_| Error::NoAvailPort)?;
     log.debug_msg(format!("bound on {}", sock.local_addr()?))
         .await;
-    upload(log, sock, targ_addr).await?;
+    upload(log, sock, tokio::io::stdin(), targ_addr).await?;
     log.debug_msg("bye!").await;
     Ok(())
 }
@@ -66,24 +68,20 @@ fn parse_args(mut args: impl Iterator<Item = String>) -> Result<SocketAddrV4, Er
 
 async fn upload(
     log: &mut Logger,
-    mut sock: tokio::net::UdpSocket,
+    mut sock: UdpSocket,
+    mut inp: Stdin,
     targ_addr: SocketAddrV4,
 ) -> Result<(), Error> {
-    let contents: String = (0..100).flat_map(|_| "Hello world ".chars()).collect();
-
-    for i in 0.. {
-        let (idx, part) = {
-            let width = 50;
-            let lo = i * width;
-            let hi = std::cmp::min((i + 1) * width, contents.len());
-            if lo >= contents.len() {
-                break;
-            };
-            (lo, &contents[lo..hi])
-        };
-        sock.send_to(part.as_bytes(), targ_addr).await?;
-        log.send_data(idx, part.len()).await;
+    let mut buf = [0u8; 500];
+    let mut pos = 0;
+    loop {
+        let len = inp.read(&mut buf).await?;
+        if len == 0 {
+            break;
+        }
+        sock.send_to(&buf[..len], targ_addr).await?;
+        log.send_data(pos, len).await;
+        pos += len;
     }
-
     Ok(())
 }