Fixed event loop scoping, removed 0.13 pyo3-asyncio API calls (#70)

Co-authored-by: Andrew J Westlake <[email protected]>

src/lib.rs

@@ -17,12 +17,11 @@ pub fn start_server() {
 }
 
 #[pymodule]
-pub fn robyn(py: Python<'_>, m: &PyModule) -> PyResult<()> {
+pub fn robyn(_py: Python<'_>, m: &PyModule) -> PyResult<()> {
     // the pymodule class to make the rustPyFunctions available
     // in python
     m.add_wrapped(wrap_pyfunction!(start_server))?;
     m.add_class::<Server>()?;
-    pyo3_asyncio::try_init(py)?;
     pyo3::prepare_freethreaded_python();
     Ok(())
 }

src/server.rs

@@ -12,8 +12,6 @@ use pyo3::prelude::*;
 use pyo3::types::PyAny;
 
 // hyper modules
-use pyo3_asyncio::run_forever;
-
 static STARTED: AtomicBool = AtomicBool::new(false);
 
 #[pyclass]
@@ -44,16 +42,29 @@ impl Server {
         let router = self.router.clone();
         let headers = self.headers.clone();
 
+        let asyncio = py.import("asyncio").unwrap();
+
+        let event_loop = asyncio.call_method0("new_event_loop").unwrap();
+        asyncio
+            .call_method1("set_event_loop", (event_loop,))
+            .unwrap();
+        let event_loop_hdl = PyObject::from(event_loop);
+
         thread::spawn(move || {
             //init_current_thread_once();
             actix_web::rt::System::new().block_on(async move {
                 let addr = format!("127.0.0.1:{}", port);
 
                 HttpServer::new(move || {
+                    let event_loop_hdl = event_loop_hdl.clone();
                     App::new()
                         .app_data(web::Data::new(router.clone()))
                         .app_data(web::Data::new(headers.clone()))
-                        .default_service(web::route().to(index))
+                        .default_service(web::route().to(move |router, headers, payload, req| {
+                            pyo3_asyncio::tokio::scope_local(event_loop_hdl.clone(), async move {
+                                index(router, headers, payload, req).await
+                            })
+                        }))
                 })
                 .bind(addr)
                 .unwrap()
@@ -63,17 +74,7 @@ impl Server {
             });
         });
 
-        // let asyncio = py.import("asyncio").unwrap();
-
-        // let event_loop = asyncio.call_method0("new_event_loop").unwrap();
-        // asyncio
-        //     .call_method1("set_event_loop", (event_loop,))
-        //     .unwrap();
-
-        // event_loop.call_method0("run_forever").unwrap();
-        // println!("test test_run_forever ... ok");
-
-        run_forever(py).unwrap()
+        event_loop.call_method0("run_forever").unwrap();
     }
 
     /// Adds a new header to our concurrent hashmap

test.py

@@ -54,4 +54,4 @@ def blocker():
 
 if __name__ == "__main__":
     app.add_header("server", "robyn")
-    app.start(port=5001)
+    app.start(port=5000)