Connect ASGI types to http_handler types

Author: Qard

src/asgi/http.rs

@@ -1,6 +1,8 @@
 use pyo3::exceptions::PyValueError;
 use pyo3::prelude::*;
 use pyo3::types::{PyAny, PyDict};
+use http_handler::{Request, RequestExt, Version};
+use std::net::SocketAddr;
 
 use crate::asgi::{AsgiInfo, HttpMethod, HttpVersion};
 
@@ -62,6 +64,88 @@ pub struct HttpConnectionScope {
   state: Option<Py<PyDict>>,
 }
 
+impl HttpConnectionScope {
+  /// Create a new HttpConnectionScope from an http::Request
+  pub fn from_request(request: &Request) -> Self {
+    // Extract HTTP version
+    let http_version = match request.version() {
+      Version::HTTP_09 => HttpVersion::V1_0, // fallback for HTTP/0.9
+      Version::HTTP_10 => HttpVersion::V1_0,
+      Version::HTTP_11 => HttpVersion::V1_1,
+      Version::HTTP_2 => HttpVersion::V2_0,
+      Version::HTTP_3 => HttpVersion::V2_0, // treat HTTP/3 as HTTP/2 for ASGI
+      _ => HttpVersion::V1_1, // default fallback
+    };
+
+    // Extract method
+    let method = HttpMethod::from(request.method().as_str());
+
+    // Extract scheme from URI or default to http
+    let scheme = request
+      .uri()
+      .scheme_str()
+      .unwrap_or("http")
+      .to_string();
+
+    // Extract path
+    let path = request.uri().path().to_string();
+
+    // Extract raw path (same as path for now, as we don't have the raw bytes)
+    let raw_path = path.clone();
+
+    // Extract query string
+    let query_string = request
+      .uri()
+      .query()
+      .unwrap_or("")
+      .to_string();
+
+    // Extract root path (default to empty)
+    let root_path = String::new();
+
+    // Convert headers
+    let headers: Vec<(String, String)> = request
+      .headers()
+      .iter()
+      .map(|(name, value)| {
+        (
+          name.as_str().to_lowercase(),
+          value.to_str().unwrap_or("").to_string(),
+        )
+      })
+      .collect();
+
+    // Extract client and server from socket info if available
+    let (client, server) = if let Some(socket_info) = request.socket_info() {
+      let client = socket_info.remote.map(|addr| match addr {
+        SocketAddr::V4(v4) => (v4.ip().to_string(), v4.port()),
+        SocketAddr::V6(v6) => (v6.ip().to_string(), v6.port()),
+      });
+      let server = socket_info.local.map(|addr| match addr {
+        SocketAddr::V4(v4) => (v4.ip().to_string(), v4.port()),
+        SocketAddr::V6(v6) => (v6.ip().to_string(), v6.port()),
+      });
+      (client, server)
+    } else {
+      (None, None)
+    };
+
+    HttpConnectionScope {
+      http_version,
+      method,
+      scheme,
+      path,
+      raw_path,
+      query_string,
+      root_path,
+      headers,
+      client,
+      server,
+      state: None,
+    }
+  }
+}
+
 impl<'py> IntoPyObject<'py> for HttpConnectionScope {
   type Target = PyDict;
   type Output = Bound<'py, Self::Target>;
@@ -233,12 +317,39 @@ impl<'py> FromPyObject<'py> for HttpSendMessage {
           })?
           .extract()?;
 
-        let headers: Vec<(String, String)> = dict
+        let headers_py = dict
           .get_item("headers")?
           .ok_or_else(|| {
             PyValueError::new_err("Missing 'headers' key in HTTP response start message")
-          })?
-          .extract()?;
+          })?;
+        
+        // Convert headers from list of lists to vec of tuples
+        let mut headers: Vec<(String, String)> = Vec::new();
+        if let Ok(headers_list) = headers_py.downcast::<pyo3::types::PyList>() {
+          for item in headers_list.iter() {
+            if let Ok(header_pair) = item.downcast::<pyo3::types::PyList>() {
+              if header_pair.len() == 2 {
+                let name = header_pair.get_item(0)?;
+                let value = header_pair.get_item(1)?;
+                
+                // Convert bytes to string
+                let name_str = if let Ok(bytes) = name.downcast::<pyo3::types::PyBytes>() {
+                  String::from_utf8_lossy(bytes.as_bytes()).to_string()
+                } else {
+                  name.extract::<String>()?
+                };
+                
+                let value_str = if let Ok(bytes) = value.downcast::<pyo3::types::PyBytes>() {
+                  String::from_utf8_lossy(bytes.as_bytes()).to_string()
+                } else {
+                  value.extract::<String>()?
+                };
+                
+                headers.push((name_str, value_str));
+              }
+            }
+          }
+        }
 
         let trailers: bool = dict
           .get_item("trailers")?

src/asgi/http_method.rs

@@ -114,6 +114,23 @@ impl From<HttpMethod> for String {
   }
 }
 
+impl From<&str> for HttpMethod {
+  fn from(method: &str) -> Self {
+    match method.to_uppercase().as_str() {
+      "GET" => HttpMethod::Get,
+      "POST" => HttpMethod::Post,
+      "PUT" => HttpMethod::Put,
+      "DELETE" => HttpMethod::Delete,
+      "PATCH" => HttpMethod::Patch,
+      "HEAD" => HttpMethod::Head,
+      "OPTIONS" => HttpMethod::Options,
+      "TRACE" => HttpMethod::Trace,
+      "CONNECT" => HttpMethod::Connect,
+      _ => HttpMethod::Get, // Default to GET for unknown methods
+    }
+  }
+}
+
 #[cfg(test)]
 mod tests {
   use super::*;

src/asgi/mod.rs

@@ -1,4 +1,7 @@
 use pyo3::prelude::*;
+use http_handler::{Request, Response};
+use bytes::BytesMut;
+use tokio::sync::oneshot;
 
 mod http;
 mod http_method;
@@ -23,45 +26,168 @@ pub use websocket::{
   WebSocketConnectionScope, WebSocketReceiveMessage, WebSocketSendException, WebSocketSendMessage,
 };
 
-fn execute_asgi_http_scope(
+pub async fn execute_asgi_http_scope(
   py_func: PyObject,
-  scope: HttpConnectionScope,
-  py: Python,
-) -> PyResult<()> {
+  request: Request,
+) -> PyResult<Response> {
+  // Create the ASGI scope from the HTTP request
+  let scope = HttpConnectionScope::from_request(&request);
+
+  // Create channels for ASGI communication
   let (rx_receiver, rx) = Receiver::http();
   let (tx_sender, mut tx) = Sender::http();
 
-  tokio::spawn(async move {
+  // Channel to receive the response data
+  let (response_tx, response_rx) = oneshot::channel::<(u16, Vec<(String, String)>, Vec<u8>)>();
+  
+  // Channel to signal Python execution completion
+  let (python_done_tx, python_done_rx) = oneshot::channel::<Result<(), String>>();
+
+  // Task to handle messages from the ASGI app
+  let response_task = tokio::spawn(async move {
+    let mut status = 500u16;
+    let mut headers = Vec::new();
+    let mut body = Vec::new();
+    let mut response_started = false;
+
     loop {
       if let Some(msg) = tx.recv().await {
-        println!("received ASGI message: {msg:#?}")
+        match msg {
+          HttpSendMessage::HttpResponseStart {
+            status: s,
+            headers: h,
+            trailers: _
+          } => {
+            status = s;
+            headers = h;
+            response_started = true;
+          }
+          HttpSendMessage::HttpResponseBody {
+            body: b,
+            more_body
+          } => {
+            if response_started {
+              body.extend_from_slice(&b);
+              if !more_body {
+                // Response is complete
+                let _ = response_tx.send((status, headers, body));
+                break;
+              }
+            }
+          }
+        }
+      } else {
+        // Channel closed
+        break;
       }
     }
   });
 
+  // Send the request body to the ASGI app
+  let request_body = request.body().clone();
   tokio::spawn(async move {
     let request_message = HttpReceiveMessage::Request {
-      body: b"Hello, world!".to_vec(),
+      body: request_body.to_vec(),
       more_body: false,
     };
 
-    rx.send(request_message).unwrap();
-
-    // Simulate disconnection after sending the request
-    let disconnect_message = HttpReceiveMessage::Disconnect;
-    rx.send(disconnect_message).unwrap();
+    if rx.send(request_message).is_err() {
+      eprintln!("Failed to send request message");
+    }
   });
 
-  let py_func = py_func.clone_ref(py);
+  // Run the ASGI app in a Python thread
+  tokio::task::spawn_blocking(move || {
+    Python::with_gil(|py| {
+      let scope_py = scope.into_pyobject(py).unwrap();
 
-  Python::with_gil(|py| {
-    if let Err(_e) = py_func.call1(
-      py,
-      (scope.into_pyobject(py).unwrap(), rx_receiver, tx_sender),
-    ) {
-      // Log or handle error
-    }
+      // Call the ASGI app to get a coroutine
+      let coroutine = match py_func.call1(py, (scope_py, rx_receiver, tx_sender)) {
+        Ok(coro) => coro,
+        Err(e) => {
+          eprintln!("Failed to call ASGI app: {}", e);
+          let _ = python_done_tx.send(Err(format!("Failed to call ASGI app: {}", e)));
+          return;
+        }
+      };
+
+      // Run the coroutine using asyncio
+      let asyncio = match py.import("asyncio") {
+        Ok(module) => module,
+        Err(e) => {
+          eprintln!("Failed to import asyncio: {}", e);
+          let _ = python_done_tx.send(Err(format!("Failed to import asyncio: {}", e)));
+          return;
+        }
+      };
+
+      // Create a new event loop for this request
+      let loop_ = match asyncio.call_method0("new_event_loop") {
+        Ok(loop_) => loop_,
+        Err(e) => {
+          eprintln!("Failed to create event loop: {}", e);
+          let _ = python_done_tx.send(Err(format!("Failed to create event loop: {}", e)));
+          return;
+        }
+      };
+
+      // Set it as the current event loop
+      if let Err(e) = asyncio.call_method1("set_event_loop", (&loop_,)) {
+        eprintln!("Failed to set event loop: {}", e);
+        let _ = python_done_tx.send(Err(format!("Failed to set event loop: {}", e)));
+        return;
+      }
+
+      // Run the coroutine
+      let result = loop_.call_method1("run_until_complete", (coroutine,));
+      
+      // Close the loop
+      let _ = loop_.call_method0("close");
+      
+      // Send the result
+      let _ = python_done_tx.send(result.map(|_| ()).map_err(|e| {
+        format!("Failed to run ASGI coroutine: {}", e)
+      }));
+    });
   });
 
-  Ok(())
+  // Wait for either the response or Python completion
+  let result = tokio::select! {
+    response = response_rx => {
+      response.map_err(|_| pyo3::exceptions::PyRuntimeError::new_err("Failed to receive response"))
+    }
+    python_result = python_done_rx => {
+      match python_result {
+        Ok(Ok(())) => {
+          // Python completed but no response was sent
+          Err(pyo3::exceptions::PyRuntimeError::new_err("ASGI app completed without sending response"))
+        }
+        Ok(Err(e)) => {
+          // Python failed with error
+          Err(pyo3::exceptions::PyRuntimeError::new_err(e))
+        }
+        Err(_) => {
+          // Channel closed unexpectedly
+          Err(pyo3::exceptions::PyRuntimeError::new_err("Python execution failed unexpectedly"))
+        }
+      }
+    }
+  };
+  
+  let (status, headers, body) = result?;
+
+  // Clean up the response task
+  response_task.abort();
+
+  // Build the HTTP response
+  let mut builder = http_handler::response::Builder::new()
+    .status(status);
+
+  for (name, value) in headers {
+    builder = builder.header(&name, &value);
+  }
+
+  builder
+    .body(BytesMut::from(&body[..]))
+    .map_err(|e| pyo3::exceptions::PyRuntimeError::new_err(format!("Failed to build response: {}", e)))
 }