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
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
// Copyright (c) 2022-2023, Radu Racariu.
use std::pin::Pin;
use tokio::sync::watch;
use libhaystack::val::{Value, kind::HaystackKind};
use uuid::Uuid;
use crate::{
base::{
Status,
input::{BaseInput, Input},
},
tokio_impl::{PinPayload, ReaderImpl, WriterImpl},
};
pub type InputImpl = BaseInput<ReaderImpl, WriterImpl>;
impl InputImpl {
pub fn new(name: &str, kind: HaystackKind, block_id: Uuid) -> Self {
// Watch is a coalescing 1-slot channel: producers always succeed and
// overwrite; consumers always see the latest value. The seed
// `(Value::Null, Status::Ok)` is what `borrow()` reports until the
// first real send.
let (writer, reader) = watch::channel::<PinPayload>((Value::Null, Status::Ok));
Self {
name: name.to_string(),
kind,
block_id,
connection_count: 0,
reader,
writer,
val: Default::default(),
status: Status::Ok,
links: Default::default(),
}
}
}
impl Input for InputImpl {
fn receiver(&mut self) -> Pin<Box<dyn Future<Output = Option<Value>> + Send + '_>> {
// Wait for the next change. We deliberately do NOT consume the value
// here: `changed()` marks the latest value as seen, so we re-arm via
// `mark_changed()` so that the subsequent `try_take()` in
// `drain_ready_inputs` actually surfaces it. Without re-arming, the
// value would be silently dropped — `read_block_inputs` discards the
// future result with `let _ = ...` and relies on `try_take` to
// surface the value.
Box::pin(async move {
match self.reader.changed().await {
Ok(()) => {
self.reader.mark_changed();
Some(self.reader.borrow().0.clone())
}
Err(_) => None,
}
})
}
fn try_take(&mut self) -> Option<(Value, Status)> {
// Watch's `has_changed` returns `Err` only if every sender was
// dropped — treat that as "no fresh value" rather than propagating.
if self.reader.has_changed().unwrap_or(false) {
Some(self.reader.borrow_and_update().clone())
} else {
None
}
}
fn set_value(&mut self, value: Value, status: Status) {
// Early-return on no-op: if neither value nor status changed
// relative to this input's cached state, every chained channel
// already has the same payload — `send_if_modified` would be a
// no-op on each, but we still pay for the value clones in the
// closure capture. Skip the loop entirely. This makes the
// semantics explicit: `set_value` only propagates on actual
// change.
if self.val.as_ref() == Some(&value) && self.status == status {
return;
}
// Forward to all chained inputs. `send_if_modified` only notifies
// when the value actually changed, so identical re-emissions don't
// wake downstream blocks — see `Output::set` for the loop-quiescence
// rationale. Status is part of the comparison so a producer's
// transition Ok→Fault wakes downstream even when the value is the
// same.
for link in &mut self.links {
if let Some(tx) = &link.tx {
tx.send_if_modified(|current| {
if current.0 != value || current.1 != status {
current.0 = value.clone();
current.1 = status;
true
} else {
false
}
});
}
}
self.val = Some(value);
self.status = status;
}
fn status(&self) -> Status {
self.status
}
}
#[cfg(test)]
mod test {
use libhaystack::val::kind::HaystackKind;
use uuid::Uuid;
use super::InputImpl;
#[test]
fn test_input_init() {
let input = InputImpl::new("test", HaystackKind::Bool, Uuid::new_v4());
assert_eq!(input.name, "test");
assert_eq!(input.kind, HaystackKind::Bool);
}
}