Core::run_background

src/reactor/mod.rs

@@ -242,6 +242,31 @@ impl Core {
         }
     }
 
+    /// Run spawned background tasks while there is at least one task registered.
+    ///
+    /// `max_wait` param specifies max time this function executes.
+    /// After specified duration this function simply returns,
+    /// and remaining background tasks kept registered.
+    ///
+    /// # Panics
+    ///
+    /// This function will **not** catch panic from spawned tasks.
+    pub fn run_background(&mut self, max_wait: Option<Duration>) {
+        let deadline = max_wait.map(|max_wait| Instant::now() + max_wait);
+        while !self.inner.borrow_mut().task_dispatch.is_empty() {
+            let turn_max_wait = if let Some(deadline) = deadline {
+                let now = Instant::now();
+                if now >= deadline {
+                    return;
+                }
+                Some(deadline - now)
+            } else {
+                None
+            };
+            self.turn(turn_max_wait);
+        }
+    }
+
     /// Performs one iteration of the event loop, blocking on waiting for events
     /// for at most `max_wait` (forever if `None`).
     ///

tests/spawn.rs

@@ -3,6 +3,9 @@ extern crate env_logger;
 extern crate futures;
 
 use std::any::Any;
+use std::sync::Arc;
+use std::sync::atomic::AtomicBool;
+use std::sync::atomic::Ordering;
 use std::sync::mpsc;
 use std::thread;
 use std::time::Duration;
@@ -145,3 +148,36 @@ fn spawn_in_drop() {
 
     lp.run(rx).unwrap();
 }
+
+#[test]
+fn run_background() {
+    let mut core = Core::new().expect("Core::new");
+
+    core.run_background(None);
+    core.run_background(Some(Duration::from_secs(100000000)));
+
+    let (tx, rx) = oneshot::channel();
+
+    let executed = Arc::new(AtomicBool::new(false));
+    let executed_copy = executed.clone();
+
+    let future = rx
+        .map(move |_| {
+            executed_copy.store(true, Ordering::SeqCst)
+        })
+        .map_err(|_| unreachable!());
+
+    core.handle().spawn(future);
+
+    core.run_background(Some(Duration::from_millis(1)));
+
+    thread::spawn(move || {
+        tx.send(1).expect("send");
+    });
+
+    core.run_background(None);
+
+    assert!(executed.load(Ordering::SeqCst));
+
+    core.run_background(None);
+}