mirror of
https://github.com/tokio-rs/tokio.git
synced 2026-08-27 00:00:12 +02:00
util: Add TokioContext future (#2791)
Co-authored-by: Lucio Franco <[email protected]>
This commit is contained in:
co-authored by
Lucio Franco
parent
262b19ae96
commit
d9d909cb4c
@@ -0,0 +1,78 @@
|
|||||||
|
//! Tokio context aware futures utilities.
|
||||||
|
//!
|
||||||
|
//! This module includes utilities around integrating tokio with other runtimes
|
||||||
|
//! by allowing the context to be attached to futures. This allows spawning
|
||||||
|
//! futures on other executors while still using tokio to drive them. This
|
||||||
|
//! can be useful if you need to use a tokio based library in an executor/runtime
|
||||||
|
//! that does not provide a tokio context.
|
||||||
|
|
||||||
|
use pin_project_lite::pin_project;
|
||||||
|
use std::{
|
||||||
|
future::Future,
|
||||||
|
pin::Pin,
|
||||||
|
task::{Context, Poll},
|
||||||
|
};
|
||||||
|
use tokio::runtime::Handle;
|
||||||
|
|
||||||
|
pin_project! {
|
||||||
|
/// `TokioContext` allows connecting a custom executor with the tokio runtime.
|
||||||
|
///
|
||||||
|
/// It contains a `Handle` to the runtime. A handle to the runtime can be
|
||||||
|
/// obtain by calling the `Runtime::handle()` method.
|
||||||
|
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))
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Trait extension that simplifies bundling a `Handle` with a `Future`.
|
||||||
|
pub trait HandleExt {
|
||||||
|
/// Convenience method that takes a Future and returns a `TokioContext`.
|
||||||
|
///
|
||||||
|
/// # Example: calling Tokio Runtime from a custom ThreadPool
|
||||||
|
///
|
||||||
|
/// ```no_run
|
||||||
|
/// use tokio_util::context::HandleExt;
|
||||||
|
/// use tokio::time::{delay_for, Duration};
|
||||||
|
///
|
||||||
|
/// let mut rt = tokio::runtime::Builder::new()
|
||||||
|
/// .threaded_scheduler()
|
||||||
|
/// .enable_all()
|
||||||
|
/// .build().unwrap();
|
||||||
|
///
|
||||||
|
/// let rt2 = tokio::runtime::Builder::new()
|
||||||
|
/// .threaded_scheduler()
|
||||||
|
/// .build().unwrap();
|
||||||
|
///
|
||||||
|
/// let fut = delay_for(Duration::from_millis(2));
|
||||||
|
///
|
||||||
|
/// rt.block_on(
|
||||||
|
/// rt2
|
||||||
|
/// .handle()
|
||||||
|
/// .wrap(async { delay_for(Duration::from_millis(2)).await }),
|
||||||
|
/// );
|
||||||
|
///```
|
||||||
|
fn wrap<F: Future>(&self, fut: F) -> TokioContext<F>;
|
||||||
|
}
|
||||||
|
|
||||||
|
impl HandleExt for Handle {
|
||||||
|
fn wrap<F: Future>(&self, fut: F) -> TokioContext<F> {
|
||||||
|
TokioContext {
|
||||||
|
inner: fut,
|
||||||
|
handle: self.clone(),
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -38,4 +38,6 @@ cfg_compat! {
|
|||||||
pub mod compat;
|
pub mod compat;
|
||||||
}
|
}
|
||||||
|
|
||||||
|
pub mod context;
|
||||||
|
|
||||||
pub mod sync;
|
pub mod sync;
|
||||||
|
|||||||
@@ -0,0 +1,28 @@
|
|||||||
|
#![warn(rust_2018_idioms)]
|
||||||
|
|
||||||
|
use tokio::runtime::Builder;
|
||||||
|
use tokio::time::*;
|
||||||
|
use tokio_util::context::HandleExt;
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn tokio_context_with_another_runtime() {
|
||||||
|
let mut rt1 = Builder::new()
|
||||||
|
.threaded_scheduler()
|
||||||
|
.core_threads(1)
|
||||||
|
// no timer!
|
||||||
|
.build()
|
||||||
|
.unwrap();
|
||||||
|
let rt2 = Builder::new()
|
||||||
|
.threaded_scheduler()
|
||||||
|
.core_threads(1)
|
||||||
|
.enable_all()
|
||||||
|
.build()
|
||||||
|
.unwrap();
|
||||||
|
|
||||||
|
// Without the `HandleExt.wrap()` there would be a panic because there is
|
||||||
|
// no timer running, since it would be referencing runtime r1.
|
||||||
|
let _ = rt1.block_on(
|
||||||
|
rt2.handle()
|
||||||
|
.wrap(async move { delay_for(Duration::from_millis(2)).await }),
|
||||||
|
);
|
||||||
|
}
|
||||||
Reference in New Issue
Block a user