Repository: actix/actix (language: rust)

Issue:
Expose TimerFunc in public API
Would it be possible to export `actix::utils::TimerFunc` to the public API just like `actix::utils::Condition`? 

I'm currently running into a case where I basically want to modify the future that is spawned by `AsyncContext::run_later` and thus would need access to it. Example code:

```rust
TimerFunc::new(duration, callback)
    // .map(), etc...
    .spawn(context);
```

My current workaround is this helper function (although I'd much prefer to use existing functionality from actix):

```rust
pub fn run_later<A, F>(
    timeout: Duration,
    f: F,
) -> impl ActorFuture<Item = (), Error = (), Actor = A>
where
    A: Actor,
    F: FnOnce(&mut A, &mut A::Context) + 'static,
{
    wrap_future(tokio_timer::Delay::new(Instant::now() + timeout)).then(|_, actor, context| {
        f(actor, context);
        fut::ok(())
    })
}

```




Discussion / hints:
It is not much different from using `Delay` I'd say, but I don't see reason not to expose it.

Though we would need to document it properly at least.
It's obviously not a big deal not to have it, but would reduce the boilerplate necessary to schedule delays and avoid to explicitly pull in `tokio-timer` as a dependency. I'm happy to piece together a PR, if you like.

Fix patch:
diff --git a/CHANGES.md b/CHANGES.md
index 6dd02d348..f47e31a15 100644
--- a/CHANGES.md
+++ b/CHANGES.md
@@ -1,5 +1,16 @@
 # CHANGES
 
+## [0.x.x](2018-1x-xx)
+
+- Introduce the `clock` module to allow overriding and mocking the system clock
+  based on `tokio_timer`.
+
+- System now has `System::builder()` which allows overriding the system clock
+  with a custom instance. `Arbiter::builder()` can now also override the system
+  clock. The default is to inherit from the system.
+
+- New utility classes `TimerFunc` and `IntervalFunc` in the `utils` module.
+
 ## [0.7.4] (2018-08-27)
 
 ### Added
diff --git a/src/actors/resolver.rs b/src/actors/resolver.rs
index bbaff9267..1d25c9cec 100644
--- a/src/actors/resolver.rs
+++ b/src/actors/resolver.rs
@@ -43,7 +43,7 @@ extern crate trust_dns_resolver;
 use std::collections::VecDeque;
 use std::io;
 use std::net::SocketAddr;
-use std::time::{Duration, Instant};
+use std::time::Duration;
 
 use self::trust_dns_resolver::config::{ResolverConfig, ResolverOpts};
 use self::trust_dns_resolver::lookup_ip::LookupIpFuture;
@@ -52,6 +52,7 @@ use futures::{Async, Future, Poll};
 use tokio_tcp::{ConnectFuture, TcpStream};
 use tokio_timer::Delay;
 
+use clock;
 use prelude::*;
 
 #[deprecated(since = "0.7.0", note = "please use `Resolver` instead")]
@@ -380,7 +381,7 @@ impl TcpConnector {
         TcpConnector {
             addrs,
             stream: None,
-            timeout: Delay::new(Instant::now() + timeout),
+            timeout: Delay::new(clock::now() + timeout),
         }
     }
 }
diff --git a/src/address/message.rs b/src/address/message.rs
index a64a7c627..e9e13f5d2 100644
--- a/src/address/message.rs
+++ b/src/address/message.rs
@@ -1,10 +1,11 @@
 use std::marker::PhantomData;
-use std::time::{Duration, Instant};
+use std::time::Duration;
 
 use futures::sync::oneshot;
 use futures::{Async, Future, Poll};
 use tokio_timer::Delay;
 
+use clock;
 use handler::{Handler, Message};
 
 use super::channel::{AddressSender, Sender};
@@ -49,7 +50,7 @@ where
 
     /// Set message delivery timeout
     pub fn timeout(mut self, dur: Duration) -> Self {
-        self.timeout = Some(Delay::new(Instant::now() + dur));
+        self.timeout = Some(Delay::new(clock::now() + dur));
         self
     }
 
@@ -130,7 +131,7 @@ where
 
     /// Set message delivery timeout
     pub fn timeout(mut self, dur: Duration) -> Self {
-        self.timeout = Some(Delay::new(Instant::now() + dur));
+        self.timeout = Some(Delay::new(clock::now() + dur));
         self
     }
 
diff --git a/src/arbiter.rs b/src/arbiter.rs
index d7bc68eb4..4a26f1cfb 100644
--- a/src/arbiter.rs
+++ b/src/arbiter.rs
@@ -5,11 +5,12 @@ use std::thread;
 use futures::sync::oneshot::{channel, Sender};
 use futures::{future, Future, IntoFuture};
 use tokio::executor::current_thread::spawn;
-use tokio::runtime::current_thread::Runtime;
+use tokio::runtime::current_thread::Builder as RuntimeBuilder;
 use uuid::Uuid;
 
 use actor::Actor;
 use address::{channel, Addr, AddressReceiver};
+use clock::Clock;
 use context::Context;
 use handler::Handler;
 use mailbox::DEFAULT_CAPACITY;
@@ -70,18 +71,18 @@ impl Arbiter {
 
     /// Spawn new thread and run event loop in spawned thread.
     /// Returns address of newly created arbiter.
-    fn new_with_builder(builder: Builder) -> Addr<Arbiter> {
+    fn new_with_builder(mut builder: Builder) -> Addr<Arbiter> {
         let (tx, rx) = std::sync::mpsc::channel();
         let id = Uuid::new_v4();
         let name = format!(
             "arbiter:{}:{}",
             id.to_hyphenated_ref().to_string(),
-            builder.name.as_ref().unwrap_or(&"actor".into())
+            builder.name.take().unwrap_or_else(|| "actor".into())
         );
         let sys = System::current();
 
         let _ = thread::Builder::new().name(name.clone()).spawn(move || {
-            let mut rt = Runtime::new().unwrap();
+            let mut rt = builder.runtime.build().unwrap();
 
             let (stop, stop_rx) = channel();
             NAME.with(|cell| *cell.borrow_mut() = Some(name));
@@ -91,8 +92,8 @@ impl Arbiter {
             System::set_current(sys);
 
             // start arbiter
-            let addr =
-                rt.block_on(future::lazy(move || {
+            let addr = rt
+                .block_on(future::lazy(move || {
                     let addr = Actor::start(Arbiter {
                         stop: Some(stop),
                         stop_system_on_panic: builder.stop_system_on_panic,
@@ -102,9 +103,10 @@ impl Arbiter {
             ADDR.with(|cell| *cell.borrow_mut() = Some(addr.clone()));
 
             // register arbiter
-            System::current()
-                .sys()
-                .do_send(RegisterArbiter(id.to_simple_ref().to_string(), addr.clone()));
+            System::current().sys().do_send(RegisterArbiter(
+                id.to_simple_ref().to_string(),
+                addr.clone(),
+            ));
 
             if tx.send(addr).is_err() {
                 error!("Can not start Arbiter, remote side is dead");
@@ -205,17 +207,17 @@ impl Arbiter {
     /// Start new arbiter and then start actor in created arbiter.
     /// Returns `Addr<Syn, A>` of created actor.
     pub fn start<A, F>(f: F) -> Addr<A>
-        where
-            A: Actor<Context = Context<A>>,
-            F: FnOnce(&mut A::Context) -> A + Send + 'static,
+    where
+        A: Actor<Context = Context<A>>,
+        F: FnOnce(&mut A::Context) -> A + Send + 'static,
     {
         Arbiter::builder().start(f)
     }
 
     fn start_with_builder<A, F>(builder: Builder, f: F) -> Addr<A>
-        where
-            A: Actor<Context = Context<A>>,
-            F: FnOnce(&mut A::Context) -> A + Send + 'static,
+    where
+        A: Actor<Context = Context<A>>,
+        F: FnOnce(&mut A::Context) -> A + Send + 'static,
     {
         let (stx, srx) = channel::channel(DEFAULT_CAPACITY);
 
@@ -271,6 +273,9 @@ pub struct Builder {
 
     /// Whether the Arbiter will stop the whole System on uncaught panic. Defaults to false.
     stop_system_on_panic: bool,
+
+    /// Tokio runtime builder.
+    runtime: RuntimeBuilder,
 }
 
 impl Builder {
@@ -278,6 +283,7 @@ impl Builder {
         Builder {
             name: None,
             stop_system_on_panic: false,
+            runtime: RuntimeBuilder::new(),
         }
     }
 
@@ -296,6 +302,14 @@ impl Builder {
         self
     }
 
+    /// Set the Clock instance that will be used by this Arbiter.
+    ///
+    /// Defaults to the clock used by the actix `System`, which defaults to the system clock.
+    pub fn clock(mut self, clock: Clock) -> Self {
+        self.runtime.clock(clock);
+        self
+    }
+
     /// Spawn new thread and run event loop in spawned thread.
     /// Returns address of newly created arbiter.
     pub fn build(self) -> Addr<Arbiter> {
@@ -305,9 +319,9 @@ impl Builder {
     /// Start new arbiter and then start actor in created arbiter.
     /// Returns `Addr<Syn, A>` of created actor.
     pub fn start<A, F>(self, f: F) -> Addr<A>
-        where
-            A: Actor<Context = Context<A>>,
-            F: FnOnce(&mut A::Context) -> A + Send + 'static,
+    where
+        A: Actor<Context = Context<A>>,
+        F: FnOnce(&mut A::Context) -> A + Send + 'static,
     {
         Arbiter::start_with_builder(self, f)
     }
diff --git a/src/clock.rs b/src/clock.rs
new file mode 100644
index 000000000..f69385a3e
--- /dev/null
+++ b/src/clock.rs
@@ -0,0 +1,11 @@
+//! A configurable source of time.
+//!
+//! This module provides an API to get the current instant in such a way that
+//! the source of time may be configured. This allows mocking out the source of
+//! time in tests.
+//!
+//! See [Module `tokio_timer::clock`] for full documentation.
+//!
+//! [Module `tokio_timer::clock`]: https://docs.rs/tokio-timer/latest/tokio_timer/clock/index.html
+
+pub use tokio_timer::clock::{now, Clock, Now};
diff --git a/src/contextitems.rs b/src/contextitems.rs
index d5ed39486..9e32769aa 100644
--- a/src/contextitems.rs
+++ b/src/contextitems.rs
@@ -1,9 +1,10 @@
 use futures::{Async, Future, Poll, Stream};
 use std::marker::PhantomData;
-use std::time::{Duration, Instant};
+use std::time::Duration;
 use tokio_timer::Delay;
 
 use actor::{Actor, ActorContext, AsyncContext};
+use clock;
 use fut::ActorFuture;
 use handler::{Handler, Message, MessageResponse};
 
@@ -57,7 +58,7 @@ where
     pub fn new(msg: M, timeout: Duration) -> Self {
         ActorDelayedMessageItem {
             msg: Some(msg),
-            timeout: Delay::new(Instant::now() + timeout),
+            timeout: Delay::new(clock::now() + timeout),
             act: PhantomData,
             m: PhantomData,
         }
diff --git a/src/fut/stream_timeout.rs b/src/fut/stream_timeout.rs
index 5a8ca0e65..6bd966e18 100644
--- a/src/fut/stream_timeout.rs
+++ b/src/fut/stream_timeout.rs
@@ -1,9 +1,10 @@
-use std::time::{Duration, Instant};
+use std::time::Duration;
 
 use futures::{Async, Future, Poll};
 use tokio_timer::Delay;
 
 use actor::Actor;
+use clock;
 use fut::ActorStream;
 
 /// Future for the `timeout` combinator, interrupts computations if it takes
@@ -57,7 +58,7 @@ where
         }
 
         if self.timeout.is_none() {
-            self.timeout = Some(Delay::new(Instant::now() + self.dur));
+            self.timeout = Some(Delay::new(clock::now() + self.dur));
         }
 
         // check timeout
diff --git a/src/fut/timeout.rs b/src/fut/timeout.rs
index c240a8c6b..a2c2cc7bd 100644
--- a/src/fut/timeout.rs
+++ b/src/fut/timeout.rs
@@ -1,9 +1,10 @@
-use std::time::{Duration, Instant};
+use std::time::Duration;
 
 use futures::{Async, Future, Poll};
 use tokio_timer::Delay;
 
 use actor::Actor;
+use clock;
 use fut::ActorFuture;
 
 /// Future for the `timeout` combinator, interrupts computations if it takes
@@ -28,7 +29,7 @@ where
     Timeout {
         fut: future,
         err: Some(err),
-        timeout: Delay::new(Instant::now() + timeout),
+        timeout: Delay::new(clock::now() + timeout),
     }
 }
 
diff --git a/src/lib.rs b/src/lib.rs
index f5dc0e53e..00c11ff71 100644
--- a/src/lib.rs
+++ b/src/lib.rs
@@ -76,6 +76,7 @@ mod address;
 mod mailbox;
 
 pub mod actors;
+pub mod clock;
 pub mod fut;
 pub mod io;
 pub mod msgs;
@@ -138,7 +139,7 @@ pub mod prelude {
         pub use msgs;
         pub use prelude::*;
         pub use registry::SystemService;
-        pub use utils::Condition;
+        pub use utils::{Condition, IntervalFunc, TimerFunc};
     }
 }
 
diff --git a/src/system.rs b/src/system.rs
index 7b85ee35e..33128652d 100644
--- a/src/system.rs
+++ b/src/system.rs
@@ -1,13 +1,15 @@
+use std::borrow::Cow;
 use std::cell::RefCell;
 use std::collections::HashMap;
 
 use futures::sync::oneshot::{channel, Receiver, Sender};
 use futures::{future, Future};
-use tokio::runtime::current_thread::Runtime;
+use tokio::runtime::current_thread::{Builder as RuntimeBuilder, Runtime};
 
 use actor::Actor;
 use address::{channel as addr_channel, Addr};
 use arbiter::Arbiter;
+use clock::Clock;
 use context::Context;
 use handler::{Handler, Message};
 use msgs::{Execute, StopArbiter};
@@ -66,12 +68,20 @@ pub struct System {
 thread_local!(static CURRENT: RefCell<Option<System>> = RefCell::new(None););
 
 impl System {
+    /// Build a new system with a customized tokio runtime.
+    ///
+    /// This allows to customize the runtime. See struct level docs on
+    /// `Builder` for more information.
+    pub fn builder() -> Builder {
+        Builder::new()
+    }
+
     #[cfg_attr(feature = "cargo-clippy", allow(new_ret_no_self))]
     /// Create new system.
     ///
     /// This method panics if it can not create tokio runtime
     pub fn new<T: Into<String>>(name: T) -> SystemRunner {
-        System::create_runtime(name.into(), || {})
+        Self::builder().name(name).build()
     }
 
     /// Get current running system.
@@ -138,44 +148,7 @@ impl System {
     where
         F: FnOnce() + 'static,
     {
-        System::create_runtime("actix".to_owned(), f).run()
-    }
-
-    fn create_runtime<F>(name: String, f: F) -> SystemRunner
-    where
-        F: FnOnce() + 'static,
-    {
-        let (stop_tx, stop) = channel();
-        let (addr_sender, addr_receiver) = addr_channel::channel(16);
-        let (arb_sender, arb_receiver) = addr_channel::channel(16);
-
-        let addr = Addr::new(arb_sender);
-        let system = System {
-            sys: Addr::new(addr_sender),
-            arbiter: addr.clone(),
-            registry: SystemRegistry::new(addr),
-        };
-        System::set_current(system.clone());
-
-        // system arbiter
-        let arb = SystemArbiter {
-            arbiters: HashMap::new(),
-            stop: Some(stop_tx),
-        };
-
-        let mut rt = Runtime::new().unwrap();
-
-        // init system arbiter and run configuration method
-        let _ = rt.block_on(future::lazy(move || {
-            Arbiter::new_system(arb_receiver, name);
-            let ctx = Context::with_receiver(addr_receiver);
-            ctx.run(arb);
-
-            f();
-            Ok::<_, ()>(())
-        }));
-
-        SystemRunner { rt, stop }
+        Self::builder().run(f)
     }
 }
 
@@ -296,3 +269,135 @@ impl<I: Send, E: Send> Handler<Execute<I, E>> for SystemArbiter {
         msg.exec()
     }
 }
+
+/// Builder struct for a System actor.
+///
+/// Either use `Builder::build` to create a system and start actors.
+/// Alternatively, use `Builder::run` to start the tokio runtime and
+/// run a function in its context.
+///
+/// # Examples
+///
+/// ```rust
+/// extern crate actix;
+///
+/// use actix::prelude::*;
+/// use actix::clock::Clock;
+/// use std::time::Duration;
+///
+/// struct Timer {
+///     dur: Duration,
+/// }
+///
+/// impl Actor for Timer {
+///     type Context = Context<Self>;
+///
+///     // stop system after `self.dur` seconds
+///     fn started(&mut self, ctx: &mut Context<Self>) {
+///         ctx.run_later(self.dur, |act, ctx| {
+///             // Stop current running system.
+///             System::current().stop();
+///         });
+///     }
+/// }
+///
+/// fn main() {
+///     // create a custom clock instance
+///     let clock = Clock::new();
+///
+///     // initialize system and run it
+///     // This function blocks current thread
+///     let code = System::builder().clock(clock).run(|| {
+///         // Start `Timer` actor. It will use the provided clock instance.
+///         Timer {
+///             dur: Duration::new(0, 1),
+///         }.start();
+///     });
+///
+///     std::process::exit(code);
+/// }
+/// ```
+pub struct Builder {
+    /// Name of the System. Defaults to "actix" if unset.
+    name: Cow<'static, str>,
+
+    /// Tokio runtime builder.
+    runtime: RuntimeBuilder,
+}
+
+impl Builder {
+    fn new() -> Self {
+        Builder {
+            name: Cow::Borrowed("actix"),
+            runtime: RuntimeBuilder::new(),
+        }
+    }
+
+    /// Sets the name of the System.
+    pub fn name<T: Into<String>>(mut self, name: T) -> Self {
+        self.name = Cow::Owned(name.into());
+        self
+    }
+
+    /// Set the Clock instance that will be used by this System.
+    ///
+    /// Defaults to the system clock.
+    pub fn clock(mut self, clock: Clock) -> Self {
+        self.runtime.clock(clock);
+        self
+    }
+
+    /// Create new System.
+    ///
+    /// This method panics if it can not create tokio runtime
+    pub fn build(self) -> SystemRunner {
+        self.create_runtime(|| {})
+    }
+
+    /// This function will start tokio runtime and will finish once the
+    /// `System::stop()` message get called.
+    /// Function `f` get called within tokio runtime context.
+    pub fn run<F>(self, f: F) -> i32
+    where
+        F: FnOnce() + 'static,
+    {
+        self.create_runtime(f).run()
+    }
+
+    fn create_runtime<F>(mut self, f: F) -> SystemRunner
+    where
+        F: FnOnce() + 'static,
+    {
+        let (stop_tx, stop) = channel();
+        let (addr_sender, addr_receiver) = addr_channel::channel(16);
+        let (arb_sender, arb_receiver) = addr_channel::channel(16);
+
+        let addr = Addr::new(arb_sender);
+        let system = System {
+            sys: Addr::new(addr_sender),
+            arbiter: addr.clone(),
+            registry: SystemRegistry::new(addr),
+        };
+        System::set_current(system.clone());
+
+        // system arbiter
+        let arb = SystemArbiter {
+            arbiters: HashMap::new(),
+            stop: Some(stop_tx),
+        };
+
+        let mut rt = self.runtime.build().unwrap();
+
+        // init system arbiter and run configuration method
+        let _ = rt.block_on(future::lazy(move || {
+            Arbiter::new_system(arb_receiver, self.name.into_owned());
+            let ctx = Context::with_receiver(addr_receiver);
+            ctx.run(arb);
+
+            f();
+            Ok::<_, ()>(())
+        }));
+
+        SystemRunner { rt, stop }
+    }
+}
diff --git a/src/utils.rs b/src/utils.rs
index b84ae614a..3a00f29a7 100644
--- a/src/utils.rs
+++ b/src/utils.rs
@@ -1,10 +1,11 @@
-use std::time::{Duration, Instant};
+use std::time::Duration;
 
 use futures::unsync::oneshot;
 use futures::{Async, Future, Poll, Stream};
 use tokio_timer::{Delay, Interval};
 
 use actor::Actor;
+use clock;
 use fut::{ActorFuture, ActorStream};
 
 pub struct Condition<T>
@@ -42,7 +43,45 @@ where
     }
 }
 
-pub(crate) struct TimerFunc<A>
+/// An `ActorFuture` that runs a function in the actor's context after a specified amount of time.
+///
+/// Unless you specifically need access to the future, use [`Context::run_later`] instead.
+///
+/// [`Context::run_later`]: ../prelude/trait.AsyncContext.html#method.run_later
+///
+/// ```rust
+/// # #[macro_use] extern crate actix;
+/// # extern crate futures;
+/// # use std::io;
+/// use std::time::Duration;
+/// use actix::prelude::*;
+/// use actix::utils::TimerFunc;
+///
+/// struct MyActor;
+///
+/// impl MyActor {
+///     fn stop(&mut self, context: &mut Context<Self>) {
+///         System::current().stop();
+///     }
+/// }
+///
+/// impl Actor for MyActor {
+///    type Context = Context<Self>;
+///
+///    fn started(&mut self, context: &mut Context<Self>) {
+///        // spawn a delayed future into our context
+///        TimerFunc::new(Duration::from_millis(100), Self::stop)
+///            .spawn(context);
+///    }
+/// }
+/// # fn main() {
+/// #    let sys = System::new("example");
+/// #    let addr = MyActor.start();
+/// #    sys.run();
+/// # }
+/// ```
+#[must_use = "future do nothing unless polled"]
+pub struct TimerFunc<A>
 where
     A: Actor,
 {
@@ -54,13 +93,14 @@ impl<A> TimerFunc<A>
 where
     A: Actor,
 {
+    /// Creates a new `TimerFunc` with the given duration.
     pub fn new<F>(timeout: Duration, f: F) -> TimerFunc<A>
     where
         F: FnOnce(&mut A, &mut A::Context) + 'static,
     {
         TimerFunc {
             f: Some(Box::new(f)),
-            timeout: Delay::new(Instant::now() + timeout),
+            timeout: Delay::new(clock::now() + timeout),
         }
     }
 }
@@ -76,7 +116,6 @@ impl<A: Actor, F: FnOnce(&mut A, &mut A::Context) + 'static> TimerFuncBox<A> for
     }
 }
 
-#[doc(hidden)]
 impl<A> ActorFuture for TimerFunc<A>
 where
     A: Actor,
@@ -86,7 +125,9 @@ where
     type Actor = A;
 
     fn poll(
-        &mut self, act: &mut Self::Actor, ctx: &mut <Self::Actor as Actor>::Context,
+        &mut self,
+        act: &mut Self::Actor,
+        ctx: &mut <Self::Actor as Actor>::Context,
     ) -> Poll<Self::Item, Self::Error> {
         match self.timeout.poll() {
             Ok(Async::Ready(_)) => {
@@ -101,19 +142,60 @@ where
     }
 }
 
-pub(crate) struct IntervalFunc<A: Actor> {
+/// An `ActorStream` that periodically runs a function in the actor's context.
+///
+/// Unless you specifically need access to the future, use [`Context::run_interval`] instead.
+///
+/// [`Context::run_interval`]: ../prelude/trait.AsyncContext.html#method.run_interval
+///
+/// ```rust
+/// # #[macro_use] extern crate actix;
+/// # extern crate futures;
+/// # use std::io;
+/// use std::time::Duration;
+/// use actix::prelude::*;
+/// use actix::utils::IntervalFunc;
+///
+/// struct MyActor;
+///
+/// impl MyActor {
+///     fn tick(&mut self, context: &mut Context<Self>) {
+///         println!("tick");
+///     }
+/// }
+///
+/// impl Actor for MyActor {
+///    type Context = Context<Self>;
+///
+///    fn started(&mut self, context: &mut Context<Self>) {
+///        // spawn an interval stream into our context
+///        IntervalFunc::new(Duration::from_millis(100), Self::tick)
+///            .finish()
+///            .spawn(context);
+/// #      context.run_later(Duration::from_millis(200), |_, _| System::current().stop());
+///    }
+/// }
+/// # fn main() {
+/// #    let sys = System::new("example");
+/// #    let addr = MyActor.start();
+/// #    sys.run();
+/// # }
+/// ```
+#[must_use = "future do nothing unless polled"]
+pub struct IntervalFunc<A: Actor> {
     f: Box<IntervalFuncBox<A>>,
     interval: Interval,
 }
 
 impl<A: Actor> IntervalFunc<A> {
+    /// Creates a new `IntervalFunc` with the given interval duration.
     pub fn new<F>(timeout: Duration, f: F) -> IntervalFunc<A>
     where
         F: FnMut(&mut A, &mut A::Context) + 'static,
     {
         Self {
             f: Box::new(f),
-            interval: Interval::new(Instant::now() + timeout, timeout),
+            interval: Interval::new(clock::now() + timeout, timeout),
         }
     }
 }
@@ -129,14 +211,15 @@ impl<A: Actor, F: FnMut(&mut A, &mut A::Context) + 'static> IntervalFuncBox<A> f
     }
 }
 
-#[doc(hidden)]
 impl<A: Actor> ActorStream for IntervalFunc<A> {
     type Item = ();
     type Error = ();
     type Actor = A;
 
     fn poll(
-        &mut self, act: &mut Self::Actor, ctx: &mut <Self::Actor as Actor>::Context,
+        &mut self,
+        act: &mut Self::Actor,
+        ctx: &mut <Self::Actor as Actor>::Context,
     ) -> Poll<Option<Self::Item>, Self::Error> {
         loop {
             match self.interval.poll() {


Test patch: