std_mel/data/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 `Map` with `key` → `value` and emit it on `map`.
7#[mel_treatment(
8    generic T ()
9    input value Block<T>
10    output map Block<Map>
11)]
12pub async fn entry(key: string) {
13    if let Ok(value) = value.recv_one().await {
14        let mut new_map = HashMap::new();
15        new_map.insert(key.clone(), value);
16        let new_map = Map { map: new_map };
17        let _ = map.send_one(Value::Data(Arc::new(new_map))).await;
18    }
19}
20
21/// Receive one `Map` block and emit the value stored under `key` as `Option<T>` on `value`.
22///
23/// Emits `none` if the key is absent or the stored value does not match type `T`.
24#[mel_treatment(
25    generic T ()
26    input map Block<Map>
27    output value Block<Option<T>>
28)]
29pub async fn get(key: string) {
30    if let Ok(map) = map.recv_one_as::<Arc<Map>>().await {
31        let _ = value.send_one_as(map.map.get(&key).cloned()).await;
32    }
33}
34
35/// Receive one `base` map and one `value` block, insert `key` → `value` into a copy of `base`, and emit the updated map on `map`.
36#[mel_treatment(
37    generic T ()
38    input base Block<Map>
39    input value Block<T>
40    output map Block<Map>
41)]
42pub async fn insert(key: string) {
43    if let (Ok(base), Ok(value)) = (base.recv_one_as::<Arc<Map>>().await, value.recv_one().await) {
44        let mut new_map = Arc::unwrap_or_clone(base);
45        new_map.map.insert(key.clone(), value);
46        let _ = map.send_one(Value::Data(Arc::new(new_map))).await;
47    }
48}
49
50/// Merge two maps
51///
52/// Merge map `entries` in `base`.
53/// `entries` erase existing entries in `base` if they already exists.
54/// `entries` can be omitted (closed input) and `merge` will still be emitted if `base` is received.
55#[mel_treatment(
56    input base Block<Map>
57    input entries Block<Map>
58    output merged Block<Map>
59)]
60pub async fn merge() {
61    if let Ok(base) = base.recv_one_as::<Arc<Map>>().await {
62        if let Ok(entries) = entries.recv_one_as::<Arc<Map>>().await {
63            let mut new_map = Arc::unwrap_or_clone(base);
64            for (key, value) in &entries.map {
65                new_map.map.insert(key.clone(), value.clone());
66            }
67
68            let _ = merged.send_one(Value::Data(Arc::new(new_map))).await;
69        } else {
70            let _ = merged.send_one(Value::Data(base)).await;
71        }
72    }
73}
std_mel/data/map/block.rs

std_mel/data/map/
block.rs
