src/main.rs

@@ -1,6 +1,5 @@
 mod git;
 mod ite;
-mod tostream;
 
 use actix_files::Files;
 use actix_web::{get, post, web, App, HttpResponse, HttpServer, Responder, Error, HttpRequest, HttpMessage};
@@ -180,6 +179,12 @@ async fn git_proto(payload : web::Payload, web::Path((owner, reponame)): web::Pa
         // future::ready(Ok(()))
     }).await;
     println!("input sent");
+    //io::copy(&mut req.take_payload(), &mut p.stdin.take().unwrap())?;
+    // Parse the headers coming out, and the pass through the rest of the
+    // process back down the stack.
+    //
+    // 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 = tokio::io::BufReader::new(p.stdout.take().unwrap());
 
     let mut headers = HashMap::new();
@@ -199,6 +204,22 @@ 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);
 
     let status_code : u16 = {
@@ -222,16 +243,56 @@ async fn git_proto(payload : web::Payload, web::Path((owner, reponame)): web::Pa
     }
 
     println!("Write body...");
+    //let mut body = Vec::new();
+    //rdr.bytes()
 
-    let unfold = stream::try_unfold(rdr, move |mut rdr| {
+    // let stream = stream::try_unfold(0, |state| async move {
+    //
+    //     if state <= 2 {
+    //         let next_state = state + 1;
+    //         let yielded = state * 2;
+    //         Ok(Some((yielded, next_state)))
+    //     } else {
+    //         Ok(None)
+    //     }
+    // });
+
+    let unfold = stream::try_unfold(0, |_useless| {
         let mut buff: [u8; 1024] = [0; 1024];
         let read = rdr.read(&mut buff[..]);
-        let result = read.map_ok(|l| Some((Bytes::copy_from_slice(&buff[0..l]), rdr)));
+        let result = read.map_ok(|l| Some((Bytes::copy_from_slice(&buff[0..l]), 0)));
         return result;
+        // match result {
+        //     Ok(l) => {Ok(Some((Bytes::copy_from_slice(&buff[0..l]), 0)))}
+        //     Err(e) => {Err(e)}
+        // }
+
+        // result.map(|l| {
+        //     print!("{}", String::from_utf8_lossy(&buff[0..l]));
+        //     Some((Bytes::copy_from_slice(&buff[0..l]), 0))
+        // });
     });
     // pin_mut!(unfold);
     let response = builder.streaming(unfold);
+
+    // let response = builder.streaming(stream::repeat_with(move || {
+    //     let mut buff : [u8; 1024] = [0; 1024];
+    //     match rdr.read(&mut buff[..]) {
+    //         Ok(l) => {
+    //             print!("{}", String::from_utf8_lossy(&buff[0..l]));
+    //             Ok(Bytes::copy_from_slice(&buff[0..l])) }
+    //         Err(e) => {Err(e)}
+    //     }
+    // })
+    //     .take_while(|bytes| {
+    //         match bytes {
+    //             Ok(bytes) => {future::ready(bytes.len() != 0)}
+    //             Err(_) => {future::ready(false)}
+    //         }
+    //     })
+    // );
     return Ok(response);
+
 }
 
 fn header(req: &HttpRequest, name: header::HeaderName) -> &str {

src/tostream.rs

@@ -1,25 +0,0 @@
-use futures::{Stream, AsyncRead};
-use std::task::{Context, Poll};
-use std::pin::Pin;
-use actix_web::web::Bytes;
-use std::io::Error;
-
-struct ToStream<T>(T);
-
-impl<T> ToStream<T> {
-    fn get_field(self: Pin<&mut Self>) -> Pin<&mut T> {
-        unsafe { self.map_unchecked_mut(|s| &mut s.0) }
-    }
-}
-
-impl <T : Unpin> Unpin for ToStream<T>{}
-
-impl <T : AsyncRead> Stream for ToStream<T>{
-    type Item = Result<Bytes, Error>;
-
-    fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
-        let mut buff: [u8; 1024] = [0; 1024];
-        let mut t = self.get_field();
-        t.poll_read(cx, &mut buff[..]).map_ok(|l| {Bytes::copy_from_slice(&buff[0..l])}).map(|res| Some(res))
-    }
-}