1use std::time::Duration;
2
3use futures::{FutureExt, StreamExt};
4use ianaio::timers::future::sleep;
5use ianaio::worker::reactor::{reactor, ReactorScope};
6
7use futures::sink::SinkExt;
8use serde::{Deserialize, Serialize};
9
10#[derive(Debug, PartialEq, Eq, Serialize, Deserialize)]
11pub enum ControlSignal {
12 Start,
13 Stop,
14}
15
16#[reactor]
17pub async fn Prime(mut scope: ReactorScope<ControlSignal, u64>) {
18 while let Some(m) = scope.next().await {
19 if m == ControlSignal::Start {
20 'inner: for i in 1.. {
21 if primes::is_prime(i) {
25 scope.send(i).await.unwrap();
26 }
27
28 futures::select! {
29 m = scope.next() => {
30 if m == Some(ControlSignal::Stop) {
31 break 'inner;
32 }
33 },
34 _ = sleep(Duration::from_millis(100)).fuse() => {},
35 }
36 }
37 }
38 }
39}
40#[cfg(test)]
51mod tests {
52 use super::*;
53
54 use ianaio::worker::Spawnable;
55 use wasm_bindgen_test::*;
56
57 wasm_bindgen_test_configure!(run_in_browser);
58
59 #[wasm_bindgen_test]
60 async fn prime_worker_works() {
61 ianaio::console::log!("running test");
62 let mut bridge = Prime::spawner().spawn("http://127.0.0.1:9999/ianaio_prime_worker.js");
63
64 bridge
65 .send(ControlSignal::Start)
66 .await
67 .expect("failed to send start signal");
68
69 sleep(Duration::from_millis(1050)).await;
70
71 bridge
72 .send(ControlSignal::Stop)
73 .await
74 .expect("failed to send stop signal");
75
76 let primes: Vec<_> = bridge.take(5).collect().await;
78 assert_eq!(primes, vec![2, 3, 5, 7, 11]);
79 }
80}