use pinned sleep instead of boxed interval
@@ -3,13 +3,18 @@ use std::pin::Pin;
use std::task::{Context, Poll};
use std::time::Duration;
+use futures_core::ready;
use pin_project_lite::pin_project;
use tokio::sync::oneshot;
use crate::actor::Actor;
-use crate::clock::{interval_at, Instant, Interval, Sleep};
+use crate::clock::{sleep, Sleep};
use crate::fut::{ActorFuture, ActorStream};
+#[deprecated(
+ since = "0.11.0",
+ note = "Please use tokio::sync::oneshot::Sender instead."
+)]
pub struct Condition<T>
where
T: Clone,
@@ -97,7 +102,7 @@ impl<A: Actor> TimerFunc<A> {
{
TimerFunc {
f: Some(Box::new(f)),
- timeout: actix_rt::time::sleep(timeout),
+ timeout: sleep(timeout),
}
}
}
@@ -115,18 +120,14 @@ where
task: &mut Context<'_>,
) -> Poll<Self::Output> {
let this = self.project();
- match this.timeout.poll(task) {
- Poll::Ready(_) => {
- if let Some(f) = this.f.take() {
+ ready!(this.timeout.poll(task));
+ let f = this.f.take().expect("TimerFunc polled after finish");
f(act, ctx);
- }
Poll::Ready(())
}
- Poll::Pending => Poll::Pending,
- }
- }
}
+pin_project! {
/// 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.
@@ -167,18 +168,22 @@ where
#[must_use = "future do nothing unless polled"]
pub struct IntervalFunc<A: Actor> {
f: Box<dyn FnMut(&mut A, &mut A::Context)>,
- interval: Interval,
+ dur: Duration,
+ #[pin]
+ timer: Sleep,
+ }
}
impl<A: Actor> IntervalFunc<A> {
/// Creates a new `IntervalFunc` with the given interval duration.
- pub fn new<F>(timeout: Duration, f: F) -> IntervalFunc<A>
+ pub fn new<F>(dur: Duration, f: F) -> IntervalFunc<A>
where
F: FnMut(&mut A, &mut A::Context) + 'static,
{
Self {
f: Box::new(f),
- interval: interval_at(Instant::now() + timeout, timeout),
+ dur,
+ timer: sleep(dur),
}
}
}
@@ -192,15 +197,12 @@ impl<A: Actor> ActorStream<A> for IntervalFunc<A> {
ctx: &mut A::Context,
task: &mut Context<'_>,
) -> Poll<Option<Self::Item>> {
- let this = self.get_mut();
+ let mut this = self.project();
loop {
- match this.interval.poll_tick(task) {
- Poll::Ready(_) => {
- //Interval Stream cannot return None
+ ready!(this.timer.as_mut().poll(task));
+ let now = this.timer.deadline();
+ this.timer.as_mut().reset(now + *this.dur);
(this.f)(act, ctx);
}
- Poll::Pending => return Poll::Pending,
- }
- }
}
}
