std_mel/data/string_map/
block.rs

1use super::*;
2use melodium_macro::mel_treatment;
3use std::collections::HashMap;
4use std::sync::Arc;
5
6/// When `value` is received, produce a single-entry `StringMap` with `key` → `value` and emit it on `map`.
7#[mel_treatment(
8    input value Block<string>
9    output map Block<StringMap>
10)]
11pub async fn entry(key: string) {
12    if let Ok(value) = value
13        .recv_one()
14        .await
15        .map(|val| GetData::<String>::try_data(val).unwrap())
16    {
17        let mut new_map = HashMap::new();
18        new_map.insert(key.clone(), value);
19        let new_map = StringMap { map: new_map };
20        let _ = map.send_one(Value::Data(Arc::new(new_map))).await;
21    }
22}
23
24/// Receive one `StringMap` block and emit the value stored under `key` as `Option<string>` on `value`.
25///
26/// Emits `none` if the key is absent.
27#[mel_treatment(
28    input map Block<StringMap>
29    output value Block<Option<string>>
30)]
31pub async fn get(key: string) {
32    if let Ok(map) = map.recv_one().await.map(|val| {
33        GetData::<Arc<dyn Data>>::try_data(val)
34            .unwrap()
35            .downcast_arc::<StringMap>()
36            .unwrap()
37    }) {
38        let _ = value.send_one(map.map.get(&key).cloned().into()).await;
39    }
40}
41
42/// Receive one `base` map and one `value` block, insert `key` → `value` into a copy of `base`, and emit the updated map on `map`.
43#[mel_treatment(
44    input base Block<StringMap>
45    input value Block<string>
46    output map Block<StringMap>
47)]
48pub async fn insert(key: string) {
49    if let (Ok(base), Ok(value)) = (
50        base.recv_one().await.map(|val| {
51            GetData::<Arc<dyn Data>>::try_data(val)
52                .unwrap()
53                .downcast_arc::<StringMap>()
54                .unwrap()
55        }),
56        value
57            .recv_one()
58            .await
59            .map(|val| GetData::<String>::try_data(val).unwrap()),
60    ) {
61        let mut new_map = Arc::unwrap_or_clone(base);
62        new_map.map.insert(key.clone(), value);
63        let _ = map.send_one(Value::Data(Arc::new(new_map))).await;
64    }
65}
66
67/// Merge two maps
68///
69/// Merge map `entries` in `base`.
70/// `entries` erase existing entries in `base` if they already exists.
71/// `entries` can be omitted (closed input) and `merge` will still be emitted if `base` is received.
72#[mel_treatment(
73    input base Block<StringMap>
74    input entries Block<StringMap>
75    output merged Block<StringMap>
76)]
77pub async fn merge() {
78    if let Ok(base) = base.recv_one().await.map(|val| {
79        GetData::<Arc<dyn Data>>::try_data(val)
80            .unwrap()
81            .downcast_arc::<StringMap>()
82            .unwrap()
83    }) {
84        if let Ok(entries) = entries.recv_one().await.map(|val| {
85            GetData::<Arc<dyn Data>>::try_data(val)
86                .unwrap()
87                .downcast_arc::<StringMap>()
88                .unwrap()
89        }) {
90            let mut new_map = Arc::unwrap_or_clone(base);
91            for (key, value) in &entries.map {
92                new_map.map.insert(key.clone(), value.clone());
93            }
94
95            let _ = merged.send_one(Value::Data(Arc::new(new_map))).await;
96        } else {
97            let _ = merged.send_one(Value::Data(base)).await;
98        }
99    }
100}
std_mel/data/string_map/block.rs

std_mel/data/string_map/
block.rs
