save

Cargo.toml

@@ -17,4 +17,4 @@ askama_actix = "0.11.1"
 env_logger = "0.8.4"
 serde = "1.0.126"
 futures = "0.3.15"
-tokio = {version = "0.2.25", features = ["time"]}
+tokio = {version = "0.2.25", features = ["time", "process"]}

src/main.rs

@@ -18,12 +18,14 @@ use crate::ite::SuperIterator;
 use std::ops::Add;
 use std::path::{PathBuf, Path};
 use serde::Deserialize;
-use std::process::{Command, Stdio, Child};
+use tokio::process::{Command, Child};
+use tokio::io::{AsyncWriteExt, AsyncBufReadExt, AsyncReadExt};
+use std::process::Stdio;
 use actix_web::http::{header, StatusCode};
 use std::io;
 use std::collections::HashMap;
 use std::io::{Read, BufRead, Write, ErrorKind};
-use futures::{StreamExt, TryStreamExt, future, stream};
+use futures::{Stream, StreamExt, TryStreamExt, future, stream, TryFutureExt};
 use actix_web::web::{Buf, Bytes};
 use actix_web::http::header::IntoHeaderValue;
 
@@ -170,8 +172,8 @@ async fn git_proto(payload : web::Payload, web::Path((owner, reponame)): web::Pa
     let mut input = p.stdin.take().unwrap();
     payload.try_for_each(|bytes| {
         // println!("{:?}", bytes);
-        input.write(bytes.bytes());
+        input.write_all(bytes.bytes()).map_err(|e| actix_web::client::PayloadError::Io(e))
-        future::ready(Ok(()))
+        // future::ready(Ok(()))
     }).await;
     println!("input sent");
     //io::copy(&mut req.take_payload(), &mut p.stdin.take().unwrap())?;
@@ -180,11 +182,12 @@ async fn git_proto(payload : web::Payload, web::Path((owner, reponame)): web::Pa
     //
     // Note that we have to be careful to not drop the process which will wait
     // for the process to exit (and we haven't read stdout)
-    let mut rdr = io::BufReader::new(p.stdout.take().unwrap());
+    let mut rdr = tokio::io::BufReader::new(p.stdout.take().unwrap());
 
     let mut headers = HashMap::new();
-    for line in rdr.by_ref().lines() {
+    while true {
-        let line = line?;
+        let mut line = String::new();
+        let len = rdr.read_line(&mut line).await?;
         if line == "" || line == "\r" {
             break;
         }
@@ -198,6 +201,21 @@ async fn git_proto(payload : web::Payload, web::Path((owner, reponame)): web::Pa
             .or_insert_with(Vec::new)
             .push(value.to_string());
     }
+    // for line in rdr.by_ref().lines() {
+    //     let line = line?;
+    //     if line == "" || line == "\r" {
+    //         break;
+    //     }
+    //
+    //     let mut parts = line.splitn(2, ':');
+    //     let key = parts.next().unwrap();
+    //     let value = parts.next().unwrap();
+    //     let value = &value[1..];
+    //     headers
+    //         .entry(key.to_string())
+    //         .or_insert_with(Vec::new)
+    //         .push(value.to_string());
+    // }
 
     println!("headers : {:?}", headers);
 
@@ -224,6 +242,16 @@ async fn git_proto(payload : web::Payload, web::Path((owner, reponame)): web::Pa
     println!("Write body...");
     //let mut body = Vec::new();
     //rdr.bytes()
+
+    let response = builder.streaming(stream::try_unfold(0, |_useless| async move  {
+        let mut buff : [u8; 1024] = [0; 1024];
+        rdr.read(&mut buff[..]).await.map(|l| {
+            print!("{}", String::from_utf8_lossy(&buff[0..l]));
+            Some(Bytes::copy_from_slice(&buff[0..l]))
+        });
+    })
+    );
+
     let response = builder.streaming(stream::repeat_with(move || {
         let mut buff : [u8; 1024] = [0; 1024];
         match rdr.read(&mut buff[..]) {
@@ -238,11 +266,8 @@ async fn git_proto(payload : web::Payload, web::Path((owner, reponame)): web::Pa
                 Ok(bytes) => {future::ready(bytes.len() != 0)}
                 Err(_) => {future::ready(false)}
             }
-        // future::ready(bytes.is_ok())
         })
     );
-    //rdr.read_to_end(&mut body)?;
-    // println!("{}", String::from_utf8(body.clone()).expect("bad utf8"));
     return Ok(response);
 
 }