mirror of https://github.com/fafhrd91/actix-net
rework type sigs.changed ExecFactory::spwan_ref name
This commit is contained in:
parent
81bbf725a6
commit
11953e13a5
|
@ -60,19 +60,19 @@ impl Default for Arbiter {
|
||||||
}
|
}
|
||||||
|
|
||||||
impl Arbiter {
|
impl Arbiter {
|
||||||
pub(crate) fn new_system<E: ExecFactory>(exec: &mut E::Executor) -> Self {
|
pub(crate) fn new_system<Exec: ExecFactory>(exec: &mut Exec::Executor) -> Self {
|
||||||
let (tx, rx) = unbounded();
|
let (tx, rx) = unbounded();
|
||||||
|
|
||||||
let arb = Arbiter::with_sender(tx);
|
let arb = Arbiter::with_sender(tx);
|
||||||
ADDR.with(|cell| *cell.borrow_mut() = Some(arb.clone()));
|
ADDR.with(|cell| *cell.borrow_mut() = Some(arb.clone()));
|
||||||
STORAGE.with(|cell| cell.borrow_mut().clear());
|
STORAGE.with(|cell| cell.borrow_mut().clear());
|
||||||
|
|
||||||
let controller: ArbiterController<E> = ArbiterController {
|
let controller = ArbiterController::<Exec> {
|
||||||
rx,
|
rx,
|
||||||
_exec: Default::default(),
|
_exec: Default::default(),
|
||||||
};
|
};
|
||||||
|
|
||||||
E::spawn_ref(exec, controller);
|
Exec::spawn_on(exec, controller);
|
||||||
|
|
||||||
arb
|
arb
|
||||||
}
|
}
|
||||||
|
@ -285,7 +285,7 @@ impl<Exec> Drop for ArbiterController<Exec> {
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
impl<E: ExecFactory> Future for ArbiterController<E> {
|
impl<Exec: ExecFactory> Future for ArbiterController<Exec> {
|
||||||
type Output = ();
|
type Output = ();
|
||||||
|
|
||||||
fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
|
fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
|
||||||
|
@ -295,7 +295,7 @@ impl<E: ExecFactory> Future for ArbiterController<E> {
|
||||||
Poll::Ready(Some(item)) => match item {
|
Poll::Ready(Some(item)) => match item {
|
||||||
ArbiterCommand::Stop => return Poll::Ready(()),
|
ArbiterCommand::Stop => return Poll::Ready(()),
|
||||||
ArbiterCommand::Execute(fut) => {
|
ArbiterCommand::Execute(fut) => {
|
||||||
E::spawn(fut);
|
Exec::spawn(fut);
|
||||||
}
|
}
|
||||||
ArbiterCommand::ExecuteFn(f) => {
|
ArbiterCommand::ExecuteFn(f) => {
|
||||||
f.call_box();
|
f.call_box();
|
||||||
|
|
|
@ -15,21 +15,21 @@ use crate::system::System;
|
||||||
/// Either use `Builder::build` to create a system and start actors.
|
/// Either use `Builder::build` to create a system and start actors.
|
||||||
/// Alternatively, use `Builder::run` to start the tokio runtime and
|
/// Alternatively, use `Builder::run` to start the tokio runtime and
|
||||||
/// run a function in its context.
|
/// run a function in its context.
|
||||||
pub struct Builder<E> {
|
pub struct Builder<Exec> {
|
||||||
/// Name of the System. Defaults to "actix" if unset.
|
/// Name of the System. Defaults to "actix" if unset.
|
||||||
name: Cow<'static, str>,
|
name: Cow<'static, str>,
|
||||||
|
|
||||||
/// Whether the Arbiter will stop the whole System on uncaught panic. Defaults to false.
|
/// Whether the Arbiter will stop the whole System on uncaught panic. Defaults to false.
|
||||||
stop_on_panic: bool,
|
stop_on_panic: bool,
|
||||||
exec: PhantomData<E>,
|
_exec: PhantomData<Exec>,
|
||||||
}
|
}
|
||||||
|
|
||||||
impl<E: ExecFactory> Builder<E> {
|
impl<Exec: ExecFactory> Builder<Exec> {
|
||||||
pub(crate) fn new() -> Builder<E> {
|
pub(crate) fn new() -> Self {
|
||||||
Builder {
|
Builder {
|
||||||
name: Cow::Borrowed("actix"),
|
name: Cow::Borrowed("actix"),
|
||||||
stop_on_panic: false,
|
stop_on_panic: false,
|
||||||
exec: PhantomData,
|
_exec: PhantomData,
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
@ -51,7 +51,7 @@ impl<E: ExecFactory> Builder<E> {
|
||||||
/// Create new System.
|
/// Create new System.
|
||||||
///
|
///
|
||||||
/// This method panics if it can not create tokio runtime
|
/// This method panics if it can not create tokio runtime
|
||||||
pub fn build(self) -> SystemRunner<E> {
|
pub fn build(self) -> SystemRunner<Exec> {
|
||||||
self.create_runtime(|| {})
|
self.create_runtime(|| {})
|
||||||
}
|
}
|
||||||
|
|
||||||
|
@ -65,8 +65,8 @@ impl<E: ExecFactory> Builder<E> {
|
||||||
self.create_runtime(f).run()
|
self.create_runtime(f).run()
|
||||||
}
|
}
|
||||||
|
|
||||||
/// Create runtime with a given instance of type that impl `ExecFactory::Executor` trait.
|
/// Create runtime with a given instance of `ExecFactory::Executor` type.
|
||||||
pub fn create_with_runtime<F>(self, mut rt: E::Executor, f: F) -> SystemRunner<E>
|
pub fn create_with_runtime<F>(self, mut rt: Exec::Executor, f: F) -> SystemRunner<Exec>
|
||||||
where
|
where
|
||||||
F: FnOnce() + 'static,
|
F: FnOnce() + 'static,
|
||||||
{
|
{
|
||||||
|
@ -75,26 +75,26 @@ impl<E: ExecFactory> Builder<E> {
|
||||||
|
|
||||||
let system = System::construct(
|
let system = System::construct(
|
||||||
sys_sender,
|
sys_sender,
|
||||||
Arbiter::new_system::<E>(&mut rt),
|
Arbiter::new_system::<Exec>(&mut rt),
|
||||||
self.stop_on_panic,
|
self.stop_on_panic,
|
||||||
);
|
);
|
||||||
|
|
||||||
// system arbiter
|
// system arbiter
|
||||||
let arb = SystemArbiter::new(stop_tx, sys_receiver);
|
let arb = SystemArbiter::new(stop_tx, sys_receiver);
|
||||||
|
|
||||||
E::spawn_ref(&mut rt, arb);
|
Exec::spawn_on(&mut rt, arb);
|
||||||
|
|
||||||
// init system arbiter and run configuration method
|
// init system arbiter and run configuration method
|
||||||
E::block_on(&mut rt, async { f() });
|
Exec::block_on(&mut rt, async { f() });
|
||||||
|
|
||||||
SystemRunner { rt, stop, system }
|
SystemRunner { rt, stop, system }
|
||||||
}
|
}
|
||||||
|
|
||||||
fn create_runtime<F>(self, f: F) -> SystemRunner<E>
|
fn create_runtime<F>(self, f: F) -> SystemRunner<Exec>
|
||||||
where
|
where
|
||||||
F: FnOnce() + 'static,
|
F: FnOnce() + 'static,
|
||||||
{
|
{
|
||||||
let rt = E::build().unwrap();
|
let rt = Exec::build().unwrap();
|
||||||
self.create_with_runtime(rt, f)
|
self.create_with_runtime(rt, f)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
@ -102,20 +102,20 @@ impl<E: ExecFactory> Builder<E> {
|
||||||
/// Helper object that runs System's event loop
|
/// Helper object that runs System's event loop
|
||||||
#[must_use = "SystemRunner must be run"]
|
#[must_use = "SystemRunner must be run"]
|
||||||
#[derive(Debug)]
|
#[derive(Debug)]
|
||||||
pub struct SystemRunner<E: ExecFactory> {
|
pub struct SystemRunner<Exec: ExecFactory> {
|
||||||
rt: E::Executor,
|
rt: Exec::Executor,
|
||||||
stop: Receiver<i32>,
|
stop: Receiver<i32>,
|
||||||
system: System,
|
system: System,
|
||||||
}
|
}
|
||||||
|
|
||||||
impl<E: ExecFactory> SystemRunner<E> {
|
impl<Exec: ExecFactory> SystemRunner<Exec> {
|
||||||
/// This function will start event loop and will finish once the
|
/// This function will start event loop and will finish once the
|
||||||
/// `System::stop()` function is called.
|
/// `System::stop()` function is called.
|
||||||
pub fn run(self) -> io::Result<()> {
|
pub fn run(self) -> io::Result<()> {
|
||||||
let SystemRunner { mut rt, stop, .. } = self;
|
let SystemRunner { mut rt, stop, .. } = self;
|
||||||
|
|
||||||
// run loop
|
// run loop
|
||||||
match E::block_on(&mut rt, stop) {
|
match Exec::block_on(&mut rt, stop) {
|
||||||
Ok(code) => {
|
Ok(code) => {
|
||||||
if code != 0 {
|
if code != 0 {
|
||||||
Err(io::Error::new(
|
Err(io::Error::new(
|
||||||
|
@ -130,11 +130,12 @@ impl<E: ExecFactory> SystemRunner<E> {
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// Spawn a future on the system arbiter.
|
||||||
pub fn spawn<F>(&mut self, fut: F)
|
pub fn spawn<F>(&mut self, fut: F)
|
||||||
where
|
where
|
||||||
F: Future<Output = ()> + 'static,
|
F: Future<Output = ()> + 'static,
|
||||||
{
|
{
|
||||||
E::spawn_ref(&mut self.rt, fut);
|
Exec::spawn_on(&mut self.rt, fut);
|
||||||
}
|
}
|
||||||
|
|
||||||
/// Execute a future and wait for result.
|
/// Execute a future and wait for result.
|
||||||
|
@ -142,6 +143,6 @@ impl<E: ExecFactory> SystemRunner<E> {
|
||||||
where
|
where
|
||||||
F: Future<Output = O>,
|
F: Future<Output = O>,
|
||||||
{
|
{
|
||||||
E::block_on(&mut self.rt, fut)
|
Exec::block_on(&mut self.rt, fut)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
|
@ -32,41 +32,23 @@ pub trait ExecFactory: Sized + Send + Sync + Unpin + 'static {
|
||||||
/// complete execution by calling `block_on` or `run`.
|
/// complete execution by calling `block_on` or `run`.
|
||||||
fn block_on<F: Future>(exec: &mut Self::Executor, f: F) -> F::Output;
|
fn block_on<F: Future>(exec: &mut Self::Executor, f: F) -> F::Output;
|
||||||
|
|
||||||
/// Spawn a future onto the single-threaded runtime without reference it.
|
/// Spawn a future onto an executor without reference it.
|
||||||
///
|
///
|
||||||
/// See [module level][mod] documentation for more details.
|
/// See [module level][mod] documentation for more details.
|
||||||
///
|
///
|
||||||
/// [mod]: index.html
|
/// [mod]: index.html
|
||||||
///
|
///
|
||||||
/// # Examples
|
|
||||||
///
|
|
||||||
/// ```rust,ignore
|
|
||||||
/// # use futures::{future, Future, Stream};
|
|
||||||
/// use actix_rt::Runtime;
|
|
||||||
///
|
|
||||||
/// # fn dox() {
|
|
||||||
/// // Create the runtime
|
|
||||||
/// let mut rt = Runtime::new().unwrap();
|
|
||||||
///
|
|
||||||
/// // Spawn a future onto the runtime
|
|
||||||
/// rt.spawn(future::lazy(|_| {
|
|
||||||
/// println!("running on the runtime");
|
|
||||||
/// }));
|
|
||||||
/// # }
|
|
||||||
/// # pub fn main() {}
|
|
||||||
/// ```
|
|
||||||
///
|
|
||||||
/// # Panics
|
/// # Panics
|
||||||
///
|
///
|
||||||
/// This function panics if the spawn fails. Failure occurs if the executor
|
/// This function panics if the spawn fails. Failure occurs if the executor
|
||||||
/// is currently at capacity and is unable to spawn a new future.
|
/// is currently at capacity and is unable to spawn a new future.
|
||||||
fn spawn<F: Future<Output = ()> + 'static>(f: F);
|
fn spawn<F: Future<Output = ()> + 'static>(f: F);
|
||||||
|
|
||||||
/// Spawn a future onto the single-threaded runtime reference. Useful when you have direct
|
/// Spawn a future onto an executor reference. Useful when you have direct
|
||||||
/// access to executor.
|
/// access to executor.
|
||||||
///
|
///
|
||||||
/// *. `spawn_ref` is preferred when you can choose between it and `spawn`.
|
/// *. `spawn_on` is preferred when you can choose between it and `spawn`.
|
||||||
fn spawn_ref<F: Future<Output = ()> + 'static>(exec: &mut Self::Executor, f: F);
|
fn spawn_on<F: Future<Output = ()> + 'static>(exec: &mut Self::Executor, f: F);
|
||||||
|
|
||||||
/// Get a timeout sleep future with given duration.
|
/// Get a timeout sleep future with given duration.
|
||||||
fn sleep(dur: Duration) -> Self::Sleep;
|
fn sleep(dur: Duration) -> Self::Sleep;
|
||||||
|
@ -105,7 +87,7 @@ impl ExecFactory for ActixExec {
|
||||||
tokio::task::spawn_local(f);
|
tokio::task::spawn_local(f);
|
||||||
}
|
}
|
||||||
|
|
||||||
fn spawn_ref<F: Future<Output = ()> + 'static>(exec: &mut Self::Executor, f: F) {
|
fn spawn_on<F: Future<Output = ()> + 'static>(exec: &mut Self::Executor, f: F) {
|
||||||
exec.1.spawn_local(f);
|
exec.1.spawn_local(f);
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
|
@ -92,7 +92,7 @@ impl ExecFactory for TokioCompatExec {
|
||||||
.unwrap();
|
.unwrap();
|
||||||
}
|
}
|
||||||
|
|
||||||
fn spawn_ref<F: Future<Output = ()> + 'static>(exec: &mut Self::Executor, f: F) {
|
fn spawn_on<F: Future<Output = ()> + 'static>(exec: &mut Self::Executor, f: F) {
|
||||||
exec.spawn_std(f);
|
exec.spawn_std(f);
|
||||||
}
|
}
|
||||||
|
|
||||||
|
@ -103,27 +103,26 @@ impl ExecFactory for TokioCompatExec {
|
||||||
|
|
||||||
#[test]
|
#[test]
|
||||||
fn tokio_compat() -> std::io::Result<()> {
|
fn tokio_compat() -> std::io::Result<()> {
|
||||||
// manually construct a compat executor.
|
let exec = TokioCompatExec::build()?;
|
||||||
let rt = TokioCompatExec::build()?;
|
|
||||||
|
|
||||||
// do some work with rt and pass it to builder
|
|
||||||
actix_rt::System::builder::<TokioCompatExec>()
|
actix_rt::System::builder::<TokioCompatExec>()
|
||||||
.create_with_runtime(rt, || {})
|
.create_with_runtime(exec, || {})
|
||||||
.block_on(async {
|
.block_on(async {
|
||||||
let (tx, rx) = tokio::sync::oneshot::channel();
|
let (tx, rx) = tokio::sync::oneshot::channel();
|
||||||
tokio_01::spawn(futures_01::lazy(|| {
|
use futures_01::Future;
|
||||||
tx.send(251).unwrap();
|
tokio_01::spawn(
|
||||||
Ok(())
|
tokio_01::timer::Delay::new(Instant::now() + Duration::from_millis(1))
|
||||||
}));
|
.map_err(|e| panic!("tokio 0.1 timer error: {}", e))
|
||||||
|
.map(|_| tx.send(251).unwrap()),
|
||||||
|
);
|
||||||
|
TokioCompatExec::sleep(Duration::from_millis(1)).await;
|
||||||
assert_eq!(251, rx.await.unwrap());
|
assert_eq!(251, rx.await.unwrap());
|
||||||
});
|
});
|
||||||
|
|
||||||
// let the system construct the executor and block on it directly.
|
|
||||||
actix_rt::System::new_with::<TokioCompatExec, _>("compat").block_on(async {
|
actix_rt::System::new_with::<TokioCompatExec, _>("compat").block_on(async {
|
||||||
let (tx, rx) = tokio::sync::oneshot::channel();
|
let (tx, rx) = tokio::sync::oneshot::channel();
|
||||||
tokio::spawn(async move {
|
tokio::spawn(async move {
|
||||||
tokio::time::delay_for(Duration::from_secs(1)).await;
|
TokioCompatExec::sleep(Duration::from_millis(1)).await;
|
||||||
tx.send(996).unwrap();
|
tx.send(996).unwrap();
|
||||||
});
|
});
|
||||||
assert_eq!(996, rx.await.unwrap());
|
assert_eq!(996, rx.await.unwrap());
|
||||||
|
|
|
@ -107,7 +107,7 @@ impl FromStream for AsyncStdTcpStream {
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
// impl trait for custom executor so server can/block_on spawn tasks
|
// impl trait for custom executor so server can block_on/spawn tasks
|
||||||
impl ExecFactory for AsyncStdExec {
|
impl ExecFactory for AsyncStdExec {
|
||||||
type Executor = ();
|
type Executor = ();
|
||||||
type Sleep = Pin<Box<dyn Future<Output = ()> + Send + 'static>>;
|
type Sleep = Pin<Box<dyn Future<Output = ()> + Send + 'static>>;
|
||||||
|
@ -124,7 +124,7 @@ impl ExecFactory for AsyncStdExec {
|
||||||
async_std::task::spawn_local(f);
|
async_std::task::spawn_local(f);
|
||||||
}
|
}
|
||||||
|
|
||||||
fn spawn_ref<F: Future<Output = ()> + 'static>(_: &mut Self::Executor, f: F) {
|
fn spawn_on<F: Future<Output = ()> + 'static>(_: &mut Self::Executor, f: F) {
|
||||||
async_std::task::spawn_local(f);
|
async_std::task::spawn_local(f);
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
|
@ -44,7 +44,7 @@ pub struct ServerBuilder<Exec = ActixExec> {
|
||||||
|
|
||||||
impl Default for ServerBuilder {
|
impl Default for ServerBuilder {
|
||||||
fn default() -> Self {
|
fn default() -> Self {
|
||||||
Self::new()
|
ServerBuilder::new()
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
@ -52,16 +52,8 @@ impl<Exec> ServerBuilder<Exec>
|
||||||
where
|
where
|
||||||
Exec: ExecFactory,
|
Exec: ExecFactory,
|
||||||
{
|
{
|
||||||
/// Create new Server builder instance with default tokio executor.
|
|
||||||
pub fn new() -> Self {
|
|
||||||
ServerBuilder::<ActixExec>::new_with()
|
|
||||||
}
|
|
||||||
|
|
||||||
/// Create new Server builder instance with a generic executor.
|
/// Create new Server builder instance with a generic executor.
|
||||||
pub fn new_with<E>() -> ServerBuilder<E>
|
pub fn new() -> Self {
|
||||||
where
|
|
||||||
E: ExecFactory,
|
|
||||||
{
|
|
||||||
let (tx, rx) = unbounded();
|
let (tx, rx) = unbounded();
|
||||||
let server = Server::new(tx);
|
let server = Server::new(tx);
|
||||||
|
|
||||||
|
|
|
@ -44,7 +44,7 @@ impl Server {
|
||||||
|
|
||||||
/// Start server building process with a custom executor
|
/// Start server building process with a custom executor
|
||||||
pub fn build_with<Exec: ExecFactory>() -> ServerBuilder<Exec> {
|
pub fn build_with<Exec: ExecFactory>() -> ServerBuilder<Exec> {
|
||||||
ServerBuilder::<Exec>::new_with()
|
ServerBuilder::<Exec>::new()
|
||||||
}
|
}
|
||||||
|
|
||||||
pub(crate) fn signal(&self, sig: Signal) {
|
pub(crate) fn signal(&self, sig: Signal) {
|
||||||
|
|
Loading…
Reference in New Issue