1use std::cell::RefCell;
25use std::future::Future;
26use std::pin::Pin;
27use std::rc::Rc;
28
29use rustdv_sim::queue::Queue;
30
31use crate::component::{Component, ComponentNode};
32use crate::port::{GetIf, PeekIf, PortName, PortOwner, PutIf, SinkHandle, bind_or_panic, sink_of};
33
34struct FifoInner<T: 'static> {
42 q: Queue<T>,
43 size: Option<usize>,
44 put_taps: RefCell<Vec<Rc<dyn SinkHandle<T>>>>,
48 get_taps: RefCell<Vec<Rc<dyn SinkHandle<T>>>>,
49}
50
51impl<T: 'static> FifoInner<T> {
52 fn new(size: Option<usize>) -> FifoInner<T> {
53 FifoInner {
54 q: match size {
55 Some(n) => Queue::new(Some(n)),
56 None => Queue::unbounded(),
57 },
58 size,
59 put_taps: RefCell::new(Vec::new()),
60 get_taps: RefCell::new(Vec::new()),
61 }
62 }
63
64 fn is_full(&self) -> bool {
65 match self.size {
66 None => false,
67 Some(s) => self.q.len() >= s,
68 }
69 }
70
71 fn tap(taps: &RefCell<Vec<Rc<dyn SinkHandle<T>>>>, item: &T) {
74 let subs: Vec<Rc<dyn SinkHandle<T>>> = taps.borrow().clone();
75 for sub in subs {
76 sub.deliver(item);
77 }
78 }
79
80 async fn put_tapped(&self, item: T) {
87 self.q.wait_for_space().await;
88 Self::tap(&self.put_taps, &item);
89 let _ = self.q.try_put(item);
90 }
91
92 fn try_put_tapped(&self, item: T) -> Result<(), T> {
93 if !self.q.has_space() {
94 return Err(item);
95 }
96 Self::tap(&self.put_taps, &item);
97 self.q.try_put(item)
98 }
99
100 fn tap_get(&self, item: Option<T>) -> Option<T> {
101 if let Some(v) = &item {
102 Self::tap(&self.get_taps, v);
103 }
104 item
105 }
106}
107
108impl<T: 'static> PutIf<T> for FifoInner<T> {
109 fn put(&self, item: T) -> Pin<Box<dyn Future<Output = ()> + '_>> {
110 Box::pin(self.put_tapped(item))
111 }
112 fn try_put(&self, item: T) -> Result<(), T> {
113 self.try_put_tapped(item)
114 }
115 fn can_put(&self) -> bool {
116 !self.is_full()
117 }
118}
119
120impl<T: 'static> GetIf<T> for FifoInner<T> {
121 fn get(&self) -> Pin<Box<dyn Future<Output = T> + '_>> {
122 Box::pin(async move {
123 let item = self.q.get().await;
124 Self::tap(&self.get_taps, &item);
125 item
126 })
127 }
128 fn try_get(&self) -> Option<T> {
129 let item = self.q.try_get();
130 self.tap_get(item)
131 }
132 fn can_get(&self) -> bool {
133 !self.q.is_empty()
134 }
135}
136
137impl<T: Clone + 'static> PeekIf<T> for FifoInner<T> {
138 fn peek(&self) -> Pin<Box<dyn Future<Output = T> + '_>> {
139 Box::pin(self.q.peek())
140 }
141 fn try_peek(&self) -> Option<T> {
142 self.q.try_peek()
143 }
144 fn can_peek(&self) -> bool {
145 !self.q.is_empty()
146 }
147}
148
149pub struct PutExport<T: 'static> {
158 iface: Rc<dyn PutIf<T>>,
159}
160
161impl<T: 'static> PutExport<T> {
162 pub fn connect(&self, owner: &dyn PortOwner, name: PortName<dyn PutIf<T>>) {
168 bind_or_panic(owner, name, self.iface.clone());
169 }
170}
171
172pub struct GetExport<T: 'static> {
174 iface: Rc<dyn GetIf<T>>,
175}
176
177impl<T: 'static> GetExport<T> {
178 pub fn connect(&self, owner: &dyn PortOwner, name: PortName<dyn GetIf<T>>) {
179 bind_or_panic(owner, name, self.iface.clone());
180 }
181}
182
183pub struct TapExport<T: 'static> {
186 taps: Rc<FifoInner<T>>,
187 on_put: bool,
188}
189
190impl<T: 'static> TapExport<T> {
191 pub fn connect(&self, owner: &dyn PortOwner, name: PortName<dyn SinkHandle<T>>) {
194 match sink_of(owner, name) {
195 Ok(sink) => {
196 let list = if self.on_put {
197 &self.taps.put_taps
198 } else {
199 &self.taps.get_taps
200 };
201 list.borrow_mut().push(sink);
202 }
203 Err(e) => panic!("{e}"),
204 }
205 }
206}
207
208pub struct PeekExport<T: 'static> {
210 iface: Rc<dyn PeekIf<T>>,
211}
212
213impl<T: 'static> PeekExport<T> {
214 pub fn connect(&self, owner: &dyn PortOwner, name: PortName<dyn PeekIf<T>>) {
215 bind_or_panic(owner, name, self.iface.clone());
216 }
217}
218
219pub struct TlmFifo<T: 'static> {
228 inner: Rc<FifoInner<T>>,
229}
230
231impl<T: 'static> Default for TlmFifo<T> {
232 fn default() -> Self {
235 TlmFifo::new(1)
236 }
237}
238
239impl<T: 'static> TlmFifo<T> {
240 pub fn new(size: usize) -> TlmFifo<T> {
243 TlmFifo {
244 inner: Rc::new(FifoInner::new(Some(size))),
245 }
246 }
247
248 pub fn unbounded() -> TlmFifo<T> {
250 TlmFifo {
251 inner: Rc::new(FifoInner::new(None)),
252 }
253 }
254
255 pub fn size(&self) -> Option<usize> {
257 self.inner.size
258 }
259 pub fn used(&self) -> usize {
261 self.inner.q.len()
262 }
263 pub fn is_empty(&self) -> bool {
264 self.inner.q.is_empty()
265 }
266 pub fn is_full(&self) -> bool {
267 self.inner.is_full()
268 }
269 pub fn flush(&self) {
271 while self.inner.q.try_get().is_some() {}
272 }
273
274 pub fn put_export(&self) -> PutExport<T> {
278 PutExport {
279 iface: self.inner.clone(),
280 }
281 }
282
283 pub fn get_export(&self) -> GetExport<T> {
285 GetExport {
286 iface: self.inner.clone(),
287 }
288 }
289
290 pub fn put_ap(&self) -> TapExport<T> {
292 TapExport {
293 taps: self.inner.clone(),
294 on_put: true,
295 }
296 }
297
298 pub fn get_ap(&self) -> TapExport<T> {
300 TapExport {
301 taps: self.inner.clone(),
302 on_put: false,
303 }
304 }
305
306 pub async fn put(&self, item: T) {
312 self.inner.put_tapped(item).await
313 }
314 pub fn try_put(&self, item: T) -> Result<(), T> {
315 self.inner.try_put_tapped(item)
316 }
317 pub async fn get(&self) -> T {
318 let item = self.inner.q.get().await;
319 FifoInner::tap(&self.inner.get_taps, &item);
320 item
321 }
322 pub fn try_get(&self) -> Option<T> {
323 let item = self.inner.q.try_get();
324 self.inner.tap_get(item)
325 }
326
327 #[cfg(test)]
330 pub(crate) fn put_iface_for_test(&self) -> Rc<dyn PutIf<T>> {
331 self.inner.clone()
332 }
333
334 pub fn handle(&self) -> TlmFifo<T> {
336 TlmFifo {
337 inner: self.inner.clone(),
338 }
339 }
340}
341
342impl<T: Clone + 'static> TlmFifo<T> {
343 pub fn peek_export(&self) -> PeekExport<T> {
346 PeekExport {
347 iface: self.inner.clone(),
348 }
349 }
350
351 pub async fn peek(&self) -> T {
352 self.inner.q.peek().await
353 }
354 pub fn try_peek(&self) -> Option<T> {
355 self.inner.q.try_peek()
356 }
357}
358
359impl<T: 'static> Component for TlmFifo<T> {}
362
363impl<T: 'static> ComponentNode for TlmFifo<T> {
364 fn node_name(&self) -> &'static str {
365 "TlmFifo"
366 }
367 fn children_mut(&mut self) -> Vec<(String, &mut (dyn ComponentNode + 'static))> {
368 Vec::new()
369 }
370}
371
372#[cfg(test)]
377mod tests {
378 use super::*;
379 use crate::port::{
380 GetPort, PeekPort, PortField, PortName, PortOwner, PutPort, SubscribePort, Subscriber,
381 };
382 use crate::shared::RustdvShared;
383 use rustdv_sim::testing::block_on;
384 use std::any::Any;
385
386 struct Holder {
389 put: PutPort<u8>,
390 get: GetPort<u8>,
391 peek: PeekPort<u8>,
392 sub: SubscribePort<u8>,
393 }
394
395 impl Holder {
396 fn new() -> Holder {
397 Holder {
398 put: PutPort::default(),
399 get: GetPort::default(),
400 peek: PeekPort::default(),
401 sub: SubscribePort::default(),
402 }
403 }
404 const PUT: PortName<dyn PutIf<u8>> = PortName::new("put");
405 const GET: PortName<dyn GetIf<u8>> = PortName::new("get");
406 const PEEK: PortName<dyn PeekIf<u8>> = PortName::new("peek");
407 const SUB: PortName<dyn SinkHandle<u8>> = PortName::new("sub");
408 }
409
410 impl PortOwner for Holder {
411 fn owner_port_slot(&self, name: &str) -> Option<Rc<dyn Any>> {
412 match name {
413 "put" => Some(self.put.slot_any()),
414 "get" => Some(self.get.slot_any()),
415 "peek" => Some(self.peek.slot_any()),
416 "sub" => Some(self.sub.slot_any()),
417 _ => None,
418 }
419 }
420 fn owner_label(&self) -> &'static str {
421 "Holder"
422 }
423 }
424
425 #[test]
426 fn a_port_is_unbound_until_connected() {
427 let h = Holder::new();
428 assert!(!h.put.bound());
429 let fifo: TlmFifo<u8> = TlmFifo::new(1);
430 fifo.put_export().connect(&h, Holder::PUT);
431 assert!(h.put.bound());
432 }
433
434 #[test]
435 fn put_and_get_through_a_fifo() {
436 block_on(async {
437 let h = Holder::new();
438 let fifo: TlmFifo<u8> = TlmFifo::new(2);
439 fifo.put_export().connect(&h, Holder::PUT);
440 fifo.get_export().connect(&h, Holder::GET);
441
442 h.put.put(1).await;
443 h.put.put(2).await;
444 assert_eq!(h.get.get().await, 1, "FIFO order through the ports");
445 assert_eq!(h.get.get().await, 2);
446 });
447 }
448
449 #[test]
450 fn peek_leaves_the_item_for_get() {
451 block_on(async {
452 let h = Holder::new();
453 let fifo: TlmFifo<u8> = TlmFifo::new(1);
454 fifo.put_export().connect(&h, Holder::PUT);
455 fifo.peek_export().connect(&h, Holder::PEEK);
456 fifo.get_export().connect(&h, Holder::GET);
457
458 h.put.put(9).await;
459 assert_eq!(h.peek.peek().await, 9);
460 assert_eq!(h.get.get().await, 9, "peek did not consume it");
461 });
462 }
463
464 #[test]
466 fn try_put_hands_a_refused_item_back() {
467 block_on(async {
468 let h = Holder::new();
469 let fifo: TlmFifo<u8> = TlmFifo::new(1);
470 fifo.put_export().connect(&h, Holder::PUT);
471 assert!(h.put.try_put(1).is_ok());
472 assert_eq!(h.put.try_put(2), Err(2), "the item comes home");
473 });
474 }
475
476 #[test]
477 fn can_put_and_can_get_track_the_fifo() {
478 block_on(async {
479 let h = Holder::new();
480 let fifo: TlmFifo<u8> = TlmFifo::new(1);
481 fifo.put_export().connect(&h, Holder::PUT);
482 fifo.get_export().connect(&h, Holder::GET);
483 assert!(h.put.can_put());
484 assert!(!h.get.can_get());
485 h.put.put(1).await;
486 assert!(!h.put.can_put());
487 assert!(h.get.can_get());
488 });
489 }
490
491 #[test]
492 fn a_bad_port_name_is_a_named_error() {
493 let h = Holder::new();
494 let nope: PortName<dyn PutIf<u8>> = PortName::new("no_such_port");
495 let fifo: TlmFifo<u8> = TlmFifo::new(1);
496 match crate::port::bind(&h, nope, fifo.put_iface_for_test()) {
497 Err(crate::port::ConnectError::NoSuchPort { owner, name }) => {
498 assert_eq!(owner, "Holder");
499 assert_eq!(name, "no_such_port");
500 }
501 other => panic!("expected NoSuchPort, got {other:?}"),
502 }
503 }
504
505 #[test]
506 fn fifo_size_used_and_flush() {
507 block_on(async {
508 let fifo: TlmFifo<u8> = TlmFifo::new(3);
509 assert_eq!(fifo.size(), Some(3));
510 assert!(fifo.is_empty());
511 fifo.put(1).await;
512 fifo.put(2).await;
513 assert_eq!(fifo.used(), 2);
514 fifo.flush();
515 assert!(fifo.is_empty(), "flush empties it");
516 });
517 }
518
519 #[test]
520 fn unbounded_is_never_full() {
521 block_on(async {
522 let fifo: TlmFifo<u8> = TlmFifo::unbounded();
523 assert_eq!(fifo.size(), None);
524 for n in 0..100 {
525 assert!(fifo.try_put(n).is_ok());
526 }
527 assert!(!fifo.is_full());
528 });
529 }
530
531 #[test]
534 fn the_put_tap_sees_every_item_and_consumes_none() {
535 #[derive(Default)]
536 struct Log {
537 seen: Vec<u8>,
538 }
539 impl Subscriber<u8> for Log {
540 fn write(&mut self, item: &u8) {
541 self.seen.push(*item);
542 }
543 }
544
545 block_on(async {
546 let h = Holder::new();
547 let log: RustdvShared<Log> = RustdvShared::default();
548 h.sub.subscribe(log.clone());
549
550 let fifo: TlmFifo<u8> = TlmFifo::unbounded();
551 fifo.put_ap().connect(&h, Holder::SUB);
552
553 for n in 1..=3u8 {
554 fifo.put(n).await;
555 }
556 assert_eq!(log.get().seen, vec![1, 2, 3], "the tap saw all three");
557 assert_eq!(fifo.used(), 3, "and took none of them");
558 });
559 }
560
561 #[test]
562 fn the_get_tap_fires_as_items_leave() {
563 #[derive(Default)]
564 struct Log {
565 seen: Vec<u8>,
566 }
567 impl Subscriber<u8> for Log {
568 fn write(&mut self, item: &u8) {
569 self.seen.push(*item);
570 }
571 }
572
573 block_on(async {
574 let h = Holder::new();
575 let log: RustdvShared<Log> = RustdvShared::default();
576 h.sub.subscribe(log.clone());
577
578 let fifo: TlmFifo<u8> = TlmFifo::unbounded();
579 fifo.get_ap().connect(&h, Holder::SUB);
580 fifo.put(7).await;
581 assert!(log.get().seen.is_empty(), "nothing has left yet");
582 let _ = fifo.get().await;
583 assert_eq!(log.get().seen, vec![7]);
584 });
585 }
586
587 #[test]
588 fn a_handle_is_the_same_fifo() {
589 block_on(async {
590 let fifo: TlmFifo<u8> = TlmFifo::unbounded();
591 let other = fifo.handle();
592 fifo.put(1).await;
593 assert_eq!(other.used(), 1, "two handles, one FIFO");
594 });
595 }
596}