fix outgoing requests

Signed-off-by: Jorge Prendes <jorge.prendes@gmail.com>

Author: jprendes

src/resource.rs

@@ -48,6 +48,7 @@ impl<T> Resource<T> {
         WriteGuard {
             guard: self.inner.clone().write_owned().await,
             resource: self.clone(),
+            do_notify: true,
         }
     }
 
@@ -56,7 +57,7 @@ impl<T> Resource<T> {
         let mut fut = pin!(fut);
         fut.as_mut().enable();
 
-        drop(guard);
+        G::unlock(guard);
 
         fut.await;
         G::lock(self.clone()).await
@@ -90,6 +91,12 @@ pub struct ReadGuard<T> {
     guard: tokio::sync::OwnedRwLockReadGuard<T>,
 }
 
+impl<T> ReadGuard<T> {
+    fn drop_no_notify(self) {
+        drop(self);
+    }
+}
+
 impl<T> Deref for ReadGuard<T> {
     type Target = T;
     fn deref(&self) -> &Self::Target {
@@ -100,6 +107,14 @@ impl<T> Deref for ReadGuard<T> {
 pub struct WriteGuard<T> {
     guard: tokio::sync::OwnedRwLockWriteGuard<T>,
     resource: Resource<T>,
+    do_notify: bool,
+}
+
+impl<T> WriteGuard<T> {
+    fn drop_no_notify(mut self) {
+        self.do_notify = false;
+        drop(self);
+    }
 }
 
 impl<T> Deref for WriteGuard<T> {
@@ -115,27 +130,36 @@ impl<T> DerefMut for WriteGuard<T> {
 }
 impl<T> Drop for WriteGuard<T> {
     fn drop(&mut self) {
-        self.resource.notify();
+        if self.do_notify {
+            self.resource.notify();
+        }
     }
 }
 
-pub trait Guard: Sized {
+trait Guard: Sized {
     type Target;
     async fn lock(res: Resource<Self::Target>) -> Self;
+    fn unlock(self);
 }
 
 impl<T> Guard for ReadGuard<T> {
     type Target = T;
     async fn lock(res: Resource<Self::Target>) -> Self {
         res.read().await
     }
+    fn unlock(self) {
+        self.drop_no_notify();
+    }
 }
 
 impl<T> Guard for WriteGuard<T> {
     type Target = T;
     async fn lock(res: Resource<Self::Target>) -> Self {
         res.write().await
     }
+    fn unlock(self) {
+        self.drop_no_notify();
+    }
 }
 
 pub trait BlockOn: Future {

src/types/http_outgoing_handler.rs

@@ -43,29 +43,29 @@ impl
         for (k, v) in headers {
             builder = builder.header(k, v);
         }
-        let body = request.body.clone();
-        let mut body = body.write_wait_until(|b| b.is_finished()).block_on();
 
-        // TODO: actually use the trailers
-        let _trailers = body.trailers.clone();
+        let future_response = Resource::new(FutureIncomingResponse::default());
+        let future_response_clone = future_response.clone();
+        async move {
+            let body = request.body.clone();
+            let mut body = body.write_wait_until(|b| b.is_finished()).await;
 
-        // TODO: use a streaming body instead of reading it all at once
-        let body = body.read_all().block_on();
+            // TODO: actually use the trailers
+            let _trailers = body.trailers.clone();
 
-        let builder = builder.body(body);
+            // TODO: use a streaming body instead of reading it all at once
+            let body = body.read_all().await;
 
-        let future_response = Resource::new(FutureIncomingResponse::default());
-        let future_response_clone = future_response.clone();
+            let builder = builder.body(body);
 
-        async move {
-            let response = builder.send().block_on();
+            let response = builder.send().await;
 
             let response = match response {
                 Ok(resp) => resp,
                 Err(err) => {
                     future_response_clone
                         .write()
-                        .block_on()
+                        .await
                         .set(Err(ErrorCode::InternalError(Some(err.to_string()))));
                     return;
                 }
@@ -80,14 +80,15 @@ impl
                 Err(err) => {
                     future_response_clone
                         .write()
-                        .block_on()
+                        .await
                         .set(Err(ErrorCode::InternalError(Some(err.to_string()))));
                     return;
                 }
             };
 
             let mut stream = Stream::new();
             let _ = stream.write(bytes);
+            let _ = stream.close();
             let body = IncomingBody {
                 stream: Resource::new(stream),
                 trailers: Resource::default(),
@@ -103,7 +104,7 @@ impl
 
             future_response_clone
                 .write()
-                .block_on()
+                .await
                 .set(Ok(Resource::new(response)));
         }
         .spawn();