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

rust_elm/core/
effect.rs
