Repository: actix/actix (language: rust)

Issue:
add function to run task every x seconds
For my use case it would be very useful to be able to run a task every x seconds,
so just like `run_later` for `AsyncContext` but instead of running once, it would repeat
every time the duration has passed.

Thank you for this useful framework!


Discussion / hints:
it is trivial to implement, is it worth to add this? new method will increase size of the api. which is large already

It seems like something that would be often useful in the context of actor systems.
It's trivial to implement but I feel like it's cleaner if it's in the API.
Akka also has a similar method in the [API](https://doc.akka.io/docs/akka/2.5/scheduler.html).
Would you provide PR?
Yeah sure :)

Fix patch:
diff --git a/src/actor.rs b/src/actor.rs
index 1a6fbb972..2c1c6bf5a 100644
--- a/src/actor.rs
+++ b/src/actor.rs
@@ -6,10 +6,10 @@ use tokio;
 use address::Addr;
 use context::Context;
 use contextitems::{ActorDelayedMessageItem, ActorMessageItem, ActorMessageStreamItem};
-use fut::ActorFuture;
+use fut::{ActorFuture, ActorStream};
 use handler::{Handler, Message};
 use stream::StreamHandler;
-use utils::TimerFunc;
+use utils::{TimerFunc, IntervalFunc};
 
 #[allow(unused_variables)]
 /// Actors are objects which encapsulate state and behavior.
@@ -386,6 +386,15 @@ where
     {
         self.spawn(TimerFunc::new(dur, f))
     }
+
+    ///Spawns job to execute closure with specified interval
+    fn run_interval<F>(&mut self, dur: Duration, f: F) -> SpawnHandle
+    where
+        F: FnMut(&mut A, &mut A::Context) + 'static,
+    {
+        self.spawn(IntervalFunc::new(dur, f).finish())
+    }
+
 }
 
 /// Spawned future handle. Could be used for cancelling spawned future.
diff --git a/src/utils.rs b/src/utils.rs
index 38b9b3a2a..7c2408147 100644
--- a/src/utils.rs
+++ b/src/utils.rs
@@ -1,11 +1,11 @@
 use std::time::{Duration, Instant};
 
 use futures::unsync::oneshot;
-use futures::{Async, Future, Poll};
-use tokio_timer::Delay;
+use futures::{Async, Future, Stream, Poll};
+use tokio_timer::{Delay, Interval};
 
 use actor::Actor;
-use fut::ActorFuture;
+use fut::{ActorFuture, ActorStream};
 
 pub struct Condition<T>
 where
@@ -100,3 +100,53 @@ where
         }
     }
 }
+
+pub(crate) struct IntervalFunc<A: Actor> {
+    f: Box<IntervalFuncBox<A>>,
+    interval: Interval,
+}
+
+impl<A: Actor> IntervalFunc<A> {
+    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),
+        }
+    }
+}
+
+trait IntervalFuncBox<A: Actor>: 'static {
+    fn call(&mut self, &mut A, &mut A::Context);
+}
+
+impl<A: Actor, F: FnMut(&mut A, &mut A::Context) + 'static> IntervalFuncBox<A> for F {
+    #[cfg_attr(feature = "cargo-clippy", allow(boxed_local))]
+    fn call(&mut self, act: &mut A, ctx: &mut A::Context) {
+        self(act, ctx)
+    }
+}
+
+#[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,
+    ) -> Poll<Option<Self::Item>, Self::Error> {
+        loop {
+            match self.interval.poll() {
+                Ok(Async::Ready(_)) => {
+                    //Interval Stream cannot return None
+                    self.f.call(act, ctx);
+                }
+                Ok(Async::NotReady) => return Ok(Async::NotReady),
+                Err(_) => unreachable!(),
+            }
+        }
+    }
+}
+


Test patch:
diff --git a/tests/test_actor.rs b/tests/test_actor.rs
index 9a6ee1145..ba5eb2b84 100644
--- a/tests/test_actor.rs
+++ b/tests/test_actor.rs
@@ -192,3 +192,39 @@ fn test_restart_sync_actor() {
     assert_eq!(stopped.load(Ordering::Relaxed), 2);
     assert_eq!(msgs.load(Ordering::Relaxed), 6);
 }
+
+struct IntervalActor(Arc<AtomicUsize>);
+
+impl Actor for IntervalActor {
+    type Context = actix::Context<Self>;
+
+    fn started(&mut self, ctx: &mut Self::Context) {
+        ctx.run_interval(Duration::from_millis(5), move |act, ctx| {
+            act.0.fetch_add(1, Ordering::Relaxed);
+
+            if act.0.load(Ordering::Relaxed) == 10 {
+                ctx.stop();
+            }
+        });
+    }
+
+    fn stopping(&mut self, _: &mut Self::Context) -> Running {
+        System::current().stop();
+        Running::Stop
+    }
+}
+
+#[test]
+fn test_run_interval() {
+    let counter = Arc::new(AtomicUsize::new(0));
+    let counter_clone = counter.clone();
+    System::run(move || {
+        let addr = IntervalActor(counter_clone).start();
+        let delay = Delay::new(Instant::now() + Duration::from_millis(1_000_000));
+        tokio::spawn(delay.then(move |_| {
+            let _addr = addr;
+            Ok(())
+        }));
+    });
+    assert_eq!(counter.load(Ordering::Relaxed), 10);
+}