src/main.rs

@@ -1,7 +1,6 @@
 mod git;
 mod ite;
-mod reader;
-mod writer;
+mod tostream;
 
 use actix_files::Files;
 use actix_web::{get, post, web, App, HttpResponse, HttpServer, Responder, Error, HttpRequest, HttpMessage};
@@ -20,8 +19,8 @@ use crate::ite::SuperIterator;
 use std::ops::Add;
 use std::path::{PathBuf, Path};
 use serde::Deserialize;
-use tokio::process::{Command, Child, ChildStdout};
-use tokio::io::{AsyncWriteExt, AsyncBufReadExt, AsyncReadExt, BufReader};
+use tokio::process::{Command, Child};
+use tokio::io::{AsyncWriteExt, AsyncBufReadExt, AsyncReadExt};
 use std::process::Stdio;
 use actix_web::http::{header, StatusCode};
 use std::io;
@@ -31,8 +30,6 @@ use futures::{Stream, StreamExt, TryStreamExt, future, stream, TryFutureExt, pin
 use actix_web::web::{Buf, Bytes};
 use actix_web::http::header::IntoHeaderValue;
 use actix_web::client::PayloadError;
-use crate::reader::ToStream;
-use crate::writer::Writer;
 
 #[derive(Template)]
 #[template(path = "hello.html")]
@@ -149,7 +146,7 @@ async fn git_main(web::Path((owner, reponame)): web::Path<(String, String)>, web
     GitMainTemplate { repo, browse : browse, root : path, user_opt : Some(user)}
 }
 //#[get("/git/{owner}/{repo}.git/{path:.*}")]
-async fn git_proto(mut payload : web::Payload, web::Path((owner, reponame)): web::Path<(String, String)>, mut req: HttpRequest) -> io::Result<HttpResponse>{
+async fn git_proto(payload : web::Payload, web::Path((owner, reponame)): web::Path<(String, String)>, mut req: HttpRequest) -> io::Result<HttpResponse>{
     println!("enter git_proto");
     let mut cmd = Command::new("git");
     cmd.arg("http-backend");
@@ -173,19 +170,23 @@ async fn git_proto(mut payload : web::Payload, web::Path((owner, reponame)): web
         .stdout(Stdio::piped())
         .stdin(Stdio::piped());
     let mut p: Child = cmd.spawn()?;
+    //p.stdin.take().unwrap().write()
     let mut input = p.stdin.take().unwrap();
-    while let Some(Ok(bytes)) = payload.next().await {
-        input.write_all(bytes.bytes()).await;
-    }
+    payload.try_for_each(|bytes| {
+        // println!("{:?}", bytes);
+        let write_all = input.write_all(bytes.bytes());
+        let res = write_all.map_err(|e| actix_web::client::PayloadError::Io(e));
+        return res;
+        // future::ready(Ok(()))
+    }).await;
     println!("input sent");
     let mut rdr = tokio::io::BufReader::new(p.stdout.take().unwrap());
 
     let mut headers = HashMap::new();
-    loop {
+    while true {
         let mut line = String::new();
         let len = rdr.read_line(&mut line).await?;
-        // println!("line : \"{}\"", line);
-        if line.len() == 2 {
+        if line == "" || line == "\r" {
             break;
         }
 
@@ -222,15 +223,14 @@ async fn git_proto(mut payload : web::Payload, web::Path((owner, reponame)): web
 
     println!("Write body...");
 
-    // let unfold = stream::try_unfold(rdr, move |mut rdr| {
-    //     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)));
-    //     return result;
-    // });
+    let unfold = stream::try_unfold(rdr, move |mut rdr| {
+        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)));
+        return result;
+    });
     // pin_mut!(unfold);
-    // let response = builder.streaming(unfold);
-    let response = builder.streaming(ToStream(rdr));
+    let response = builder.streaming(unfold);
     return Ok(response);
 }
 

src/tostream.rs

@@ -1,10 +1,10 @@
-// use futures::{Stream, AsyncRead, TryStream};
+use futures::{Stream, AsyncRead};
 use std::task::{Context, Poll};
 use std::pin::Pin;
 use actix_web::web::Bytes;
 use std::io::Error;
 
-pub struct ToStream<T>(pub T);
+struct ToStream<T>(T);
 
 impl<T> ToStream<T> {
     fn get_field(self: Pin<&mut Self>) -> Pin<&mut T> {
@@ -14,7 +14,7 @@ impl<T> ToStream<T> {
 
 impl <T : Unpin> Unpin for ToStream<T>{}
 
-impl <T : tokio::io::AsyncRead> futures::Stream 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>> {

src/writer.rs

@@ -1,13 +0,0 @@
-use actix_web::web::{Bytes, Buf};
-use tokio::io::AsyncWriteExt;
-use actix_web::error::PayloadError;
-
-pub struct Writer<T>(pub T);
-
-impl <T : tokio::io::AsyncWrite + Unpin> Writer<T> {
-    pub async fn write(&mut self, bytes : Bytes) -> Result<(), PayloadError> {
-        let write_all = self.0.write_all(bytes.bytes()).await;
-        let res = write_all.map_err(|e| actix_web::client::PayloadError::Io(e));
-        return res;
-    }
-}