discro 0.35.0

DISCRete Observables for asynchronous Functional Reactive Programming (FRP).
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51

// SPDX-FileCopyrightText: The discro authors
// SPDX-License-Identifier: MPL-2.0

use std::panic;

use crate::{OrphanedSubscriberError, Subscriber};

#[allow(dead_code, reason = "Unused by default features.")]
pub(crate) async fn map_changed<S, T>(
    subscriber: &mut Subscriber<S>,
    mut map_fn: impl FnMut(&S) -> T,
) -> Result<T, OrphanedSubscriberError> {
    let next_changed = subscriber.read_changed().await?;
    let result = panic::catch_unwind(panic::AssertUnwindSafe(|| map_fn(&next_changed)));
    match result {
        Ok(next_item) => Ok(next_item),
        Err(panicked) => {
            // Drop the read-lock to avoid poisoning it.
            drop(next_changed);
            // Forward the panic to the caller.
            panic::resume_unwind(panicked);
            // Unreachable
        }
    }
}

#[allow(dead_code, reason = "Unused by default features.")]
pub(crate) async fn filter_map_changed<S, T>(
    subscriber: &mut Subscriber<S>,
    mut filter_map_fn: impl FnMut(&S) -> Option<T>,
) -> Result<T, OrphanedSubscriberError> {
    loop {
        let next_changed = subscriber.read_changed().await?;
        let result = panic::catch_unwind(panic::AssertUnwindSafe(|| filter_map_fn(&next_changed)));
        match result {
            Ok(Some(next_item)) => {
                return Ok(next_item);
            }
            Ok(None) => {
                // Continue.
            }
            Err(panicked) => {
                // Drop the read-lock to avoid poisoning it.
                drop(next_changed);
                // Forward the panic to the caller.
                panic::resume_unwind(panicked);
                // Unreachable
            }
        }
    }
}