test_issue.rs

#[pyclass]
struct TestIssue {
    rt: tokio::runtime::Runtime,
}

#[pymethods]
impl TestIssue {
    #[new]
    fn new() -> Self {

        let mut rt = tokio::runtime::Builder::new()
                            .enable_all()
                            .basic_scheduler()
                            .build()
                            .expect("build runtime");

        TestIssue {
            rt
        }
    }

    fn start_server<'f>(&mut self, _py: Python, _loop_: PyObject) -> PyResult<()> {
        //let (fut, res_fut): (PyObject, PyObject) = {
        //    let fut: PyObject = self.loop_.call_method0(py, "create_future")?.into();
        //   (fut.clone_ref(py), fut)
        //};

        self.start_runtime_and_server();

        Ok(())
    }


}

impl TestIssue {
    fn start_runtime_and_server(&mut self) {
        let handle = self.rt.handle().clone();

        thread::spawn(move || {
            let local = tokio::task::LocalSet::new();
            handle.block_on(local.run_until(async move {
                let addr = ([0, 0, 0, 0], 8000).into();

                // Using a !Send request counter is fine on 1 thread...
                let counter: Rc<Cell<usize>> = Rc::new(Cell::new(0));

                let make_service = make_service_fn(move |_| {

                    // For each connection, clone the counter to use in our service...
                    let cnt = counter.clone();

                    async move {
                        Ok::<_, Error>(service_fn(move |_| {
                            let prev = cnt.get();
                            cnt.set(prev + 1);
                            let value = cnt.get();

                            async move {
                                Ok::<_, Error>(Response::new(Body::from(format!("Request #{}", value))))
                            }
                        }))
                    }
                });

                let server = Server::bind(&addr)
                    .executor(CustomExecutor)
                    .serve(make_service);

                println!("Listening on http://{}", addr);

                // The server would block on current thread to await !Send futures.
                if let Err(e) = server.await {
                    eprintln!("server error: {}", e);
                };
            }));
        });

    }
}


#[derive(Clone, Copy, Debug)]
struct CustomExecutor;

impl<F> hyper::rt::Executor<F> for CustomExecutor
where
    F: std::future::Future + 'static, // not requiring `Send`
{
    fn execute(&self, fut: F) {
        // This will spawn into the currently running `LocalSet`.
        tokio::task::spawn_local(fut);
    }
}