util: Add TokioContext future

tokio-util/Cargo.toml

@@ -42,10 +42,11 @@ log = "0.4"
 pin-project-lite = "0.1.4"
 
 [dev-dependencies]
+lazy_static = "1.4.0"
 tokio = { version = "0.3.0", path = "../tokio", features = ["full"] }
 tokio-test = { version = "0.3.0", path = "../tokio-test" }
 
-futures = "0.3.0"
+futures = { version = "0.3.0", features = ["thread-pool"] }
 futures-test = "0.3.5"
 
 [package.metadata.docs.rs]

tokio-util/src/context.rs

@@ -0,0 +1,38 @@
+//! This module introduces `TokioContext`, which enables a simple way of using
+//! the tokio runtime with any other executor.
+
+use pin_project_lite::pin_project;
+use std::{
+    future::Future,
+    pin::Pin,
+    task::{Context, Poll},
+};
+use tokio::runtime::Handle;
+
+pin_project! {
+    /// TODO: add docs. Get some inspiration?
+    pub struct TokioContext<F> {
+        #[pin]
+        inner: F,
+        handle: Handle,
+    }
+}
+
+impl<F: Future> Future for TokioContext<F> {
+    type Output = F::Output;
+    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
+        let me = self.project();
+        let handle = me.handle;
+        let fut = me.inner;
+
+        handle.enter(|| fut.poll(cx))
+    }
+}
+
+impl<F: Future> TokioContext<F> {
+    /// Creates a new `TokioContext`. Expects a future as its first argument, and a
+    /// `tokio::runtime::Handle` as a second.
+    pub fn new(f: F, handle: Handle) -> Self {
+        Self { inner: f, handle }
+    }
+}

tokio-util/src/lib.rs

@@ -38,4 +38,6 @@ cfg_compat! {
     pub mod compat;
 }
 
+pub mod context;
+
 pub mod sync;

tokio-util/tests/context.rs

@@ -0,0 +1,49 @@
+#![warn(rust_2018_idioms)]
+
+use tokio::{net::TcpListener, sync::oneshot};
+use tokio_util::context::TokioContext;
+
+use lazy_static::lazy_static;
+use std::future::Future;
+
+struct ThreadPool {
+    inner: futures::executor::ThreadPool,
+    rt: tokio::runtime::Runtime,
+}
+
+lazy_static! {
+    static ref EXECUTOR: ThreadPool = {
+        // Spawn tokio runtime on a single background thread
+        // enabling IO and timers.
+        let rt = tokio::runtime::Builder::new()
+            .basic_scheduler()
+            .enable_all()
+            .core_threads(1)
+            .build()
+            .unwrap();
+
+        let inner = futures::executor::ThreadPool::builder().create().unwrap();
+
+        ThreadPool { inner, rt }
+    };
+}
+
+impl ThreadPool {
+    fn spawn(&self, f: impl Future<Output = ()> + Send + 'static) {
+        let handle = self.rt.handle().clone();
+        self.inner.spawn_ok(TokioContext::new(f, handle));
+    }
+}
+
+#[test]
+fn tokio_context_with_another_runtime() {
+    let (tx, rx) = oneshot::channel();
+
+    EXECUTOR.spawn(async move {
+        let listener = TcpListener::bind("0.0.0.0:0").await.unwrap();
+        println!("addr: {:?}", listener.local_addr());
+        tx.send(()).unwrap();
+    });
+
+    futures::executor::block_on(rx).unwrap();
+}