rust_elm/
effect.rs

1use std::any::Any;
2use std::collections::HashMap;
3use std::future::Future;
4use std::pin::Pin;
5use std::sync::atomic::{AtomicU64, Ordering};
6use std::sync::{Arc, Mutex, OnceLock};
7use std::time::Duration;
8
9use crate::bus::BusSender;
10use rust_dependencies::DependencyValues;
11use crate::env::Environment;
12use crate::error::EffectError;
13
14pub type EffectId = u64;
15
16type ErasedTask = Arc<
17    dyn Fn() -> Pin<Box<dyn Future<Output = Result<Box<dyn Any + Send>, EffectError>> + Send>>
18        + Send
19        + Sync,
20>;
21
22type ErasedEnvTask = Arc<
23    dyn Fn(
24            &Environment,
25        ) -> Pin<Box<dyn Future<Output = Result<Box<dyn Any + Send>, EffectError>> + Send>>
26        + Send
27        + Sync,
28>;
29
30type ErasedRun = Arc<
31    dyn Fn(
32            Box<dyn Any + Send>,
33        ) -> Pin<Box<dyn Future<Output = Result<(), EffectError>> + Send>>
34        + Send
35        + Sync,
36>;
37
38static TASK_REGISTRY: OnceLock<Mutex<HashMap<EffectId, ErasedTask>>> = OnceLock::new();
39static ENV_TASK_REGISTRY: OnceLock<Mutex<HashMap<EffectId, ErasedEnvTask>>> = OnceLock::new();
40static RUN_REGISTRY: OnceLock<Mutex<HashMap<EffectId, ErasedRun>>> = OnceLock::new();
41static NEXT_REGISTRY_ID: AtomicU64 = AtomicU64::new(1);
42
43fn task_registry() -> &'static Mutex<HashMap<EffectId, ErasedTask>> {
44    TASK_REGISTRY.get_or_init(|| Mutex::new(HashMap::new()))
45}
46
47fn env_task_registry() -> &'static Mutex<HashMap<EffectId, ErasedEnvTask>> {
48    ENV_TASK_REGISTRY.get_or_init(|| Mutex::new(HashMap::new()))
49}
50
51fn run_registry() -> &'static Mutex<HashMap<EffectId, ErasedRun>> {
52    RUN_REGISTRY.get_or_init(|| Mutex::new(HashMap::new()))
53}
54
55/// Emitter passed to [`Effect::from_run`] closures (UDF `.run { send in … }`).
56pub struct RunSender<M> {
57    pub(crate) tx: BusSender<M>,
58}
59
60impl<M> RunSender<M> {
61    pub fn send(&self, msg: M) {
62        let _ = self.tx.send_blocking(msg);
63    }
64}
65
66pub(crate) fn register_task<M, F>(run: F) -> EffectId
67where
68    M: Send + 'static,
69    F: Fn() -> Pin<Box<dyn Future<Output = Result<M, EffectError>> + Send>> + Send + Sync + 'static,
70{
71    let id = NEXT_REGISTRY_ID.fetch_add(1, Ordering::Relaxed);
72    task_registry().lock().unwrap().insert(
73        id,
74        Arc::new(move || {
75            let fut = run();
76            Box::pin(async move {
77                fut.await.map(|value| Box::new(value) as Box<dyn Any + Send>)
78            })
79        }),
80    );
81    id
82}
83
84pub(crate) fn register_env_task<M, F>(run: F) -> EffectId
85where
86    M: Send + 'static,
87    F: Fn(&Environment) -> Pin<Box<dyn Future<Output = Result<M, EffectError>> + Send>>
88        + Send
89        + Sync
90        + 'static,
91{
92    let id = NEXT_REGISTRY_ID.fetch_add(1, Ordering::Relaxed);
93    env_task_registry().lock().unwrap().insert(
94        id,
95        Arc::new(move |env| {
96            let fut = run(env);
97            Box::pin(async move {
98                fut.await.map(|value| Box::new(value) as Box<dyn Any + Send>)
99            })
100        }),
101    );
102    id
103}
104
105pub(crate) fn run_registered_task<M: Send + 'static>(
106    id: EffectId,
107) -> Pin<Box<dyn Future<Output = Result<M, EffectError>> + Send>> {
108    let registry = task_registry().lock().unwrap();
109    let task = registry.get(&id).expect("missing registered task").clone();
110    drop(registry);
111    Box::pin(async move {
112        let any = (task)().await?;
113        any.downcast::<M>()
114            .map(|boxed| *boxed)
115            .map_err(|_| EffectError::TaskFailed("task message type mismatch"))
116    })
117}
118
119pub(crate) fn run_registered_env_task<M: Send + 'static>(
120    env: &Environment,
121    id: EffectId,
122) -> Pin<Box<dyn Future<Output = Result<M, EffectError>> + Send>> {
123    let registry = env_task_registry().lock().unwrap();
124    let task = registry.get(&id).expect("missing registered env task").clone();
125    let env = env.clone();
126    drop(registry);
127    Box::pin(async move {
128        let any = (task)(&env).await?;
129        any.downcast::<M>()
130            .map(|boxed| *boxed)
131            .map_err(|_| EffectError::TaskFailed("env task message type mismatch"))
132    })
133}
134
135pub(crate) fn register_run<M, F>(run: F) -> EffectId
136where
137    M: Send + 'static,
138    F: Fn(RunSender<M>) -> Pin<Box<dyn Future<Output = Result<(), EffectError>> + Send>>
139        + Send
140        + Sync
141        + 'static,
142{
143    let id = NEXT_REGISTRY_ID.fetch_add(1, Ordering::Relaxed);
144    run_registry().lock().unwrap().insert(
145        id,
146        Arc::new(move |any: Box<dyn Any + Send>| {
147            let tx = *any
148                .downcast::<BusSender<M>>()
149                .expect("run sender type mismatch");
150            run(RunSender { tx })
151        }),
152    );
153    id
154}
155
156pub(crate) fn run_registered_run<M: Send + 'static>(
157    id: EffectId,
158    tx: BusSender<M>,
159) -> Pin<Box<dyn Future<Output = Result<(), EffectError>> + Send>> {
160    let registry = run_registry().lock().unwrap();
161    let run = registry.get(&id).expect("missing registered run").clone();
162    drop(registry);
163    Box::pin(async move { (run)(Box::new(tx) as Box<dyn Any + Send>).await })
164}
165
166/// Async work descriptor — fn pointer at the description boundary.
167pub type TaskFn<M> = fn() -> Pin<Box<dyn Future<Output = Result<M, EffectError>> + Send>>;
168
169pub type EnvTaskFn<M> =
170    fn(&Environment) -> Pin<Box<dyn Future<Output = Result<M, EffectError>> + Send>>;
171
172/// Pure effect descriptions — interpreted only in `runtime.rs`.
173///
174/// - [`Effect::merge`] / [`Effect::batch`] — run children concurrently (UDF merge).
175/// - [`Effect::concatenate`] / [`Effect::sequence`] — run children in order (UDF concatenate).
176pub enum Effect<M> {
177    None,
178    Task { id: EffectId, run: TaskFn<M> },
179    RegisteredTask { id: EffectId },
180    EnvTask { id: EffectId, run: EnvTaskFn<M> },
181    RegisteredEnvTask { id: EffectId },
182    RegisteredRun { id: EffectId },
183    Batch(Vec<Effect<M>>),
184    Cancellable {
185        id: EffectId,
186        cancel_in_flight: bool,
187        inner: Box<Effect<M>>,
188    },
189    Debounce {
190        id: EffectId,
191        duration: Duration,
192        inner: Box<Effect<M>>,
193    },
194    Throttle {
195        id: EffectId,
196        duration: Duration,
197        latest: bool,
198        inner: Box<Effect<M>>,
199    },
200    Provide {
201        env: Environment,
202        inner: Box<Effect<M>>,
203    },
204    Retry {
205        attempts: u32,
206        inner: Box<Effect<M>>,
207    },
208    Timeout {
209        duration: Duration,
210        inner: Box<Effect<M>>,
211    },
212    Sequence(Vec<Effect<M>>),
213    Race(Vec<Effect<M>>),
214    Catch {
215        inner: Box<Effect<M>>,
216        recover: fn(EffectError) -> Effect<M>,
217    },
218    Cancel { id: EffectId },
219}
220
221impl<M> Clone for Effect<M> {
222    fn clone(&self) -> Self {
223        match self {
224            Self::None => Self::None,
225            Self::Task { id, run } => Self::Task { id: *id, run: *run },
226            Self::RegisteredTask { id } => Self::RegisteredTask { id: *id },
227            Self::EnvTask { id, run } => Self::EnvTask { id: *id, run: *run },
228            Self::RegisteredEnvTask { id } => Self::RegisteredEnvTask { id: *id },
229            Self::RegisteredRun { id } => Self::RegisteredRun { id: *id },
230            Self::Batch(items) => Self::Batch(items.clone()),
231            Self::Cancellable {
232                id,
233                cancel_in_flight,
234                inner,
235            } => Self::Cancellable {
236                id: *id,
237                cancel_in_flight: *cancel_in_flight,
238                inner: inner.clone(),
239            },
240            Self::Debounce {
241                id,
242                duration,
243                inner,
244            } => Self::Debounce {
245                id: *id,
246                duration: *duration,
247                inner: inner.clone(),
248            },
249            Self::Throttle {
250                id,
251                duration,
252                latest,
253                inner,
254            } => Self::Throttle {
255                id: *id,
256                duration: *duration,
257                latest: *latest,
258                inner: inner.clone(),
259            },
260            Self::Provide { env, inner } => Self::Provide {
261                env: env.clone(),
262                inner: inner.clone(),
263            },
264            Self::Retry { attempts, inner } => Self::Retry {
265                attempts: *attempts,
266                inner: inner.clone(),
267            },
268            Self::Timeout { duration, inner } => Self::Timeout {
269                duration: *duration,
270                inner: inner.clone(),
271            },
272            Self::Sequence(items) => Self::Sequence(items.clone()),
273            Self::Race(items) => Self::Race(items.clone()),
274            Self::Catch { inner, recover } => Self::Catch {
275                inner: inner.clone(),
276                recover: *recover,
277            },
278            Self::Cancel { id } => Self::Cancel { id: *id },
279        }
280    }
281}
282
283impl<M> Effect<M> {
284    pub fn none() -> Self {
285        Self::None
286    }
287
288    pub fn task(id: EffectId, run: TaskFn<M>) -> Self {
289        Self::Task { id, run }
290    }
291
292    pub fn from_fn<F>(run: F) -> Self
293    where
294        M: Send + 'static,
295        F: Fn() -> Pin<Box<dyn Future<Output = Result<M, EffectError>> + Send>>
296            + Send
297            + Sync
298            + 'static,
299    {
300        let id = register_task(run);
301        Self::RegisteredTask { id }
302    }
303
304    pub fn env_task(id: EffectId, run: EnvTaskFn<M>) -> Self {
305        Self::EnvTask { id, run }
306    }
307
308    pub fn from_env_fn<F>(run: F) -> Self
309    where
310        M: Send + 'static,
311        F: Fn(&Environment) -> Pin<Box<dyn Future<Output = Result<M, EffectError>> + Send>>
312            + Send
313            + Sync
314            + 'static,
315    {
316        let id = register_env_task(run);
317        Self::RegisteredEnvTask { id }
318    }
319
320    pub fn from_run<F>(run: F) -> Self
321    where
322        M: Send + 'static,
323        F: Fn(RunSender<M>) -> Pin<Box<dyn Future<Output = Result<(), EffectError>> + Send>>
324            + Send
325            + Sync
326            + 'static,
327    {
328        let id = register_run(run);
329        Self::RegisteredRun { id }
330    }
331
332    /// Single-shot async task (alias for [`Effect::task`]).
333    pub fn task_try(id: EffectId, run: TaskFn<M>) -> Self {
334        Self::task(id, run)
335    }
336
337    /// Map a `Result<T, E>` leaf task into `M` via fn pointers (UDF `TaskResult`).
338    pub fn result_task<T, E>(run: TaskFn<Result<T, E>>, on_ok: fn(T) -> M, on_err: fn(E) -> M) -> Self
339    where
340        T: Send + 'static,
341        E: Send + 'static,
342        M: Send + 'static,
343    {
344        Self::from_fn(move || {
345            let fut = run();
346            Box::pin(async move {
347                match fut.await {
348                    Ok(Ok(value)) => Ok(on_ok(value)),
349                    Ok(Err(err)) => Ok(on_err(err)),
350                    Err(effect_err) => Err(effect_err),
351                }
352            })
353        })
354    }
355
356    pub fn batch(effects: impl IntoIterator<Item = Effect<M>>) -> Self {
357        let effects: Vec<_> = effects.into_iter().collect();
358        if effects.is_empty() {
359            Self::None
360        } else if effects.len() == 1 {
361            effects.into_iter().next().unwrap()
362        } else {
363            Self::Batch(effects)
364        }
365    }
366
367    pub fn merge(effects: impl IntoIterator<Item = Effect<M>>) -> Self {
368        Self::batch(effects)
369    }
370
371    pub fn concatenate(effects: impl IntoIterator<Item = Effect<M>>) -> Self {
372        let effects: Vec<_> = effects.into_iter().collect();
373        if effects.is_empty() {
374            Self::None
375        } else {
376            Self::Sequence(effects)
377        }
378    }
379
380    pub fn map<N>(self, f: fn(M) -> N) -> Effect<N>
381    where
382        M: Send + 'static,
383        N: Send + 'static,
384    {
385        match self {
386            Self::None => Effect::None,
387            Self::Task { run, .. } => Effect::from_fn(move || {
388                let fut = run();
389                Box::pin(async move { fut.await.map(f) })
390            }),
391            Self::RegisteredTask { id } => Effect::from_fn(move || {
392                let fut = run_registered_task::<M>(id);
393                Box::pin(async move { fut.await.map(f) })
394            }),
395            Self::EnvTask { run, .. } => Effect::from_env_fn(move |env| {
396                let fut = run(env);
397                Box::pin(async move { fut.await.map(f) })
398            }),
399            Self::RegisteredEnvTask { id } => Effect::from_env_fn(move |env| {
400                let fut = run_registered_env_task::<M>(env, id);
401                Box::pin(async move { fut.await.map(f) })
402            }),
403            Self::RegisteredRun { id } => Effect::RegisteredRun { id },
404            Self::Batch(items) => Effect::Batch(items.into_iter().map(|e| e.map(f)).collect()),
405            Self::Cancellable {
406                id,
407                cancel_in_flight,
408                inner,
409            } => Effect::Cancellable {
410                id,
411                cancel_in_flight,
412                inner: Box::new(inner.map(f)),
413            },
414            Self::Debounce {
415                id,
416                duration,
417                inner,
418            } => Effect::Debounce {
419                id,
420                duration,
421                inner: Box::new(inner.map(f)),
422            },
423            Self::Throttle {
424                id,
425                duration,
426                latest,
427                inner,
428            } => Effect::Throttle {
429                id,
430                duration,
431                latest,
432                inner: Box::new(inner.map(f)),
433            },
434            Self::Provide { env, inner } => Effect::Provide {
435                env,
436                inner: Box::new(inner.map(f)),
437            },
438            Self::Retry { attempts, inner } => Effect::Retry {
439                attempts,
440                inner: Box::new(inner.map(f)),
441            },
442            Self::Timeout { duration, inner } => Effect::Timeout {
443                duration,
444                inner: Box::new(inner.map(f)),
445            },
446            Self::Sequence(items) => Effect::Sequence(items.into_iter().map(|e| e.map(f)).collect()),
447            Self::Race(items) => Effect::Race(items.into_iter().map(|e| e.map(f)).collect()),
448            Self::Catch { inner, recover: _ } => inner.map(f),
449            Self::Cancel { id } => Effect::Cancel { id },
450        }
451    }
452
453    pub fn cancellable(id: EffectId, inner: Effect<M>) -> Self {
454        Self::cancellable_with(id, true, inner)
455    }
456
457    pub fn cancellable_with(id: EffectId, cancel_in_flight: bool, inner: Effect<M>) -> Self {
458        Self::Cancellable {
459            id,
460            cancel_in_flight,
461            inner: Box::new(inner),
462        }
463    }
464
465    pub fn debounce(id: EffectId, duration: Duration, inner: Effect<M>) -> Self {
466        Self::Debounce {
467            id,
468            duration,
469            inner: Box::new(inner),
470        }
471    }
472
473    pub fn throttle(id: EffectId, duration: Duration, latest: bool, inner: Effect<M>) -> Self {
474        Self::Throttle {
475            id,
476            duration,
477            latest,
478            inner: Box::new(inner),
479        }
480    }
481
482    pub fn cancel(id: EffectId) -> Self {
483        Self::Cancel { id }
484    }
485
486    pub fn provide(env: Environment, inner: Effect<M>) -> Self {
487        Self::Provide {
488            env,
489            inner: Box::new(inner),
490        }
491    }
492
493    /// Scoped single-dependency override (UDF dependency override / `withDependencies`).
494    pub fn provide_dependency<D: Send + Sync + 'static>(value: D, inner: Effect<M>) -> Self {
495        Self::provide(Environment::from_values(DependencyValues::new().with(value)), inner)
496    }
497
498    pub fn retry(attempts: u32, inner: Effect<M>) -> Self {
499        Self::Retry {
500            attempts,
501            inner: Box::new(inner),
502        }
503    }
504
505    pub fn timeout(duration: Duration, inner: Effect<M>) -> Self {
506        Self::Timeout {
507            duration,
508            inner: Box::new(inner),
509        }
510    }
511
512    pub fn sequence(effects: impl IntoIterator<Item = Effect<M>>) -> Self {
513        Self::concatenate(effects)
514    }
515
516    pub fn race(effects: impl IntoIterator<Item = Effect<M>>) -> Self {
517        let effects: Vec<_> = effects.into_iter().collect();
518        if effects.is_empty() {
519            Self::None
520        } else {
521            Self::Race(effects)
522        }
523    }
524
525    pub fn catch(inner: Effect<M>, recover: fn(EffectError) -> Effect<M>) -> Self {
526        Self::Catch {
527            inner: Box::new(inner),
528            recover,
529        }
530    }
531}
532
533impl<M> PartialEq for Effect<M> {
534    fn eq(&self, other: &Self) -> bool {
535        std::mem::discriminant(self) == std::mem::discriminant(other)
536    }
537}
538
539impl<M> std::fmt::Debug for Effect<M> {
540    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
541        match self {
542            Self::None => write!(f, "Effect::None"),
543            Self::Task { id, .. } => write!(f, "Effect::Task({id})"),
544            Self::RegisteredTask { id } => write!(f, "Effect::RegisteredTask({id})"),
545            Self::EnvTask { id, .. } => write!(f, "Effect::EnvTask({id})"),
546            Self::RegisteredEnvTask { id } => write!(f, "Effect::RegisteredEnvTask({id})"),
547            Self::RegisteredRun { id } => write!(f, "Effect::RegisteredRun({id})"),
548            Self::Batch(n) => write!(f, "Effect::Batch({})", n.len()),
549            Self::Cancellable {
550                id,
551                cancel_in_flight,
552                ..
553            } => write!(f, "Effect::Cancellable({id}, in_flight={cancel_in_flight})"),
554            Self::Debounce { id, duration, .. } => {
555                write!(f, "Effect::Debounce({id}, {duration:?})")
556            }
557            Self::Throttle {
558                id,
559                duration,
560                latest,
561                ..
562            } => write!(f, "Effect::Throttle({id}, {duration:?}, latest={latest})"),
563            Self::Provide { .. } => write!(f, "Effect::Provide"),
564            Self::Retry { attempts, .. } => write!(f, "Effect::Retry({attempts})"),
565            Self::Timeout { duration, .. } => write!(f, "Effect::Timeout({duration:?})"),
566            Self::Sequence(n) => write!(f, "Effect::Sequence({})", n.len()),
567            Self::Race(n) => write!(f, "Effect::Race({})", n.len()),
568            Self::Catch { .. } => write!(f, "Effect::Catch"),
569            Self::Cancel { id } => write!(f, "Effect::Cancel({id})"),
570        }
571    }
572}
573
574pub(crate) fn run_leaf<M>(
575    effect: Effect<M>,
576    env: &Environment,
577) -> Pin<Box<dyn Future<Output = Result<M, EffectError>> + Send>>
578where
579    M: Send + 'static,
580{
581    match effect {
582        Effect::Task { run, .. } => run(),
583        Effect::RegisteredTask { id } => run_registered_task(id),
584        Effect::EnvTask { run, .. } => run(env),
585        Effect::RegisteredEnvTask { id } => run_registered_env_task(env, id),
586        Effect::Cancel { .. } => Box::pin(async move {
587            Err(EffectError::Cancelled)
588        }),
589        other => Box::pin(async move {
590            Err(EffectError::Other(format!("non-leaf effect: {other:?}")))
591        }),
592    }
593}
594
595#[cfg(test)]
596#[allow(dead_code, clippy::redundant_closure, clippy::type_complexity)]
597mod tests {
598    use super::*;
599
600    #[test]
601    fn merge_and_concatenate_constructors() {
602        let a = Effect::<i32>::none();
603        let b = Effect::<i32>::none();
604        assert!(matches!(Effect::merge([a, b]), Effect::Batch(_)));
605        assert!(matches!(Effect::concatenate([] as [Effect<i32>; 0]), Effect::None));
606    }
607
608    #[test]
609    fn result_task_maps_ok_and_err() {
610        fn load() -> Pin<Box<dyn Future<Output = Result<Result<i32, &'static str>, EffectError>> + Send>> {
611            Box::pin(async { Ok(Ok(7)) })
612        }
613        fn fail() -> Pin<Box<dyn Future<Output = Result<Result<i32, &'static str>, EffectError>> + Send>> {
614            Box::pin(async { Ok(Err("nope")) })
615        }
616        fn ok(n: i32) -> String {
617            format!("ok:{n}")
618        }
619        fn err(e: &'static str) -> String {
620            format!("err:{e}")
621        }
622
623        let ok_effect = Effect::result_task(load, ok, err);
624        let err_effect = Effect::result_task(fail, ok, err);
625        assert!(matches!(ok_effect, Effect::RegisteredTask { .. }));
626        assert!(matches!(err_effect, Effect::RegisteredTask { .. }));
627    }
628
629    #[test]
630    fn debounce_and_throttle_constructors() {
631        let inner = Effect::<i32>::task(1, || Box::pin(async { Ok(1) }));
632        assert!(matches!(
633            Effect::debounce(9, Duration::from_millis(50), inner.clone()),
634            Effect::Debounce { id: 9, .. }
635        ));
636        assert!(matches!(
637            Effect::throttle(9, Duration::from_millis(50), true, inner),
638            Effect::Throttle {
639                id: 9,
640                latest: true,
641                ..
642            }
643        ));
644    }
645}
rust_elm/effect.rs

rust_elm/
effect.rs
