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
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
use std::sync::{
Arc,
atomic::{AtomicBool, AtomicUsize, Ordering},
};
use parking_lot::Mutex;
use super::{CellValue, Watchable};
use crate::{
cell::{Cell, CellImmutable, CellMutable},
signal::Signal,
traits::Mutable,
};
/// Combines a vector of cells into a single cell that emits `Vec<T>`.
///
/// The resulting cell emits whenever any input cell changes.
/// Completes when all input cells complete.
/// Errors immediately if any input cell errors.
///
/// # Example
/// ```
/// use hyphae::{Cell, Mutable, Gettable, join_vec};
///
/// let a = Cell::new(1);
/// let b = Cell::new(2);
/// let c = Cell::new(3);
///
/// let combined = join_vec(vec![a.clone().lock(), b.lock(), c.lock()]);
/// assert_eq!(combined.get(), vec![1, 2, 3]);
///
/// a.set(10);
/// assert_eq!(combined.get(), vec![10, 2, 3]);
/// ```
#[track_caller]
pub fn join_vec<T, W>(cells: Vec<W>) -> Cell<Vec<T>, CellImmutable>
where
T: CellValue,
W: Watchable<T> + Clone + Send + Sync + 'static,
{
let caller = std::panic::Location::caller();
if cells.is_empty() {
let derived = Cell::<Vec<T>, CellMutable>::new(vec![]);
derived.complete();
return derived
.with_name(format!(
"join_vec@{}:{}:{}[0]",
caller.file(),
caller.line(),
caller.column()
))
.lock();
}
// Get initial values
let initial: Vec<T> = cells.iter().map(|c| c.get()).collect();
// Shared "last known values", one slot per cell, held through the ENTIRE
// update-then-notify sequence below (not just the read that builds the
// combined vec). This matters only under the `scheduler` feature's
// wave-parallel draining, where multiple same-height cells can now
// notify at the literal same instant on different threads: each cell's
// callback used to re-read every cell's `.get()` independently, so
// whichever notify's *push* into the scheduler's coalescing slot landed
// last could carry a stale peek at a sibling that hadn't updated yet,
// leaving a torn combined vec as the survivor (same class of bug
// confirmed on `join`'s two-cell version by repro). Holding this lock
// across the notify call, not just the read, guarantees whichever
// cell's push actually lands last also reflects the freshest state of
// every cell — no sibling update can land in the gap between "I read
// the combined vec" and "I pushed it".
let latest: Arc<Mutex<Vec<T>>> = Arc::new(Mutex::new(initial.clone()));
let derived = Cell::<Vec<T>, CellMutable>::new(initial);
let join_name = if let Some(name) = cells.first().and_then(|c| c.name()) {
format!(
"{}::join_vec@{}:{}:{}[{}]",
name,
caller.file(),
caller.line(),
caller.column(),
num_cells_from(&cells)
)
} else {
format!(
"join_vec@{}:{}:{}[{}]",
caller.file(),
caller.line(),
caller.column(),
num_cells_from(&cells)
)
};
let derived = derived.with_name(join_name);
let num_cells = cells.len();
let complete_count = Arc::new(AtomicUsize::new(0));
// Subscribe to each cell
for (i, cell) in cells.iter().enumerate() {
let weak = derived.downgrade();
let first = Arc::new(AtomicBool::new(true));
let cc = complete_count.clone();
let nc = num_cells;
let latest = latest.clone();
let guard = cell.subscribe(move |signal| {
if let Some(d) = weak.upgrade() {
match signal {
Signal::Value(v) => {
// Skip first emission (initial value already set)
if first.swap(false, Ordering::SeqCst) {
return;
}
let mut guard = latest.lock();
guard[i] = v.as_ref().clone();
d.notify(Signal::value(guard.clone()));
}
Signal::Complete => {
let prev = cc.fetch_add(1, Ordering::SeqCst);
if prev + 1 == nc {
// All cells have completed
d.notify(Signal::Complete);
}
}
Signal::Error(e) => {
// Error from any cell propagates immediately
d.notify(Signal::Error(e.clone()));
}
}
}
});
derived.own(guard);
}
derived.lock()
}
fn num_cells_from<T, W>(cells: &[W]) -> usize
where
T: CellValue,
W: Watchable<T> + Clone + Send + Sync + 'static,
{
cells.len()
}
#[cfg(test)]
mod tests {
use super::*;
use crate::{Mutable, traits::Gettable};
#[test]
fn test_join_vec_empty() {
let combined: Cell<Vec<i32>, CellImmutable> =
join_vec::<i32, Cell<i32, CellImmutable>>(vec![]);
assert_eq!(combined.get(), Vec::<i32>::new());
assert!(combined.is_complete());
}
#[test]
fn test_join_vec_single() {
let a = Cell::new(42);
let a_locked = a.clone().lock();
let combined = join_vec(vec![a_locked]);
assert_eq!(combined.get(), vec![42]);
a.set(100);
assert_eq!(combined.get(), vec![100]);
}
#[test]
fn test_join_vec_multiple() {
let a = Cell::new(1);
let b = Cell::new(2);
let c = Cell::new(3);
let combined = join_vec(vec![a.clone().lock(), b.clone().lock(), c.clone().lock()]);
assert_eq!(combined.get(), vec![1, 2, 3]);
a.set(10);
assert_eq!(combined.get(), vec![10, 2, 3]);
b.set(20);
assert_eq!(combined.get(), vec![10, 20, 3]);
c.set(30);
assert_eq!(combined.get(), vec![10, 20, 30]);
}
#[test]
fn test_join_vec_completion() {
let a = Cell::new(1);
let b = Cell::new(2);
let combined = join_vec(vec![a.clone().lock(), b.clone().lock()]);
assert!(!combined.is_complete());
a.complete();
assert!(!combined.is_complete());
b.complete();
assert!(combined.is_complete());
}
#[test]
fn test_join_vec_subscription() {
use std::sync::atomic::AtomicI32;
let a = Cell::new(1);
let b = Cell::new(2);
let combined = join_vec(vec![a.clone().lock(), b.clone().lock()]);
let count = Arc::new(AtomicI32::new(0));
let count_clone = count.clone();
let _guard = combined.subscribe(move |signal| {
if let Signal::Value(_) = signal {
count_clone.fetch_add(1, Ordering::SeqCst);
}
});
// Initial subscription triggers once
assert_eq!(count.load(Ordering::SeqCst), 1);
a.set(10);
assert_eq!(count.load(Ordering::SeqCst), 2);
b.set(20);
assert_eq!(count.load(Ordering::SeqCst), 3);
}
}