streamweave-attractor 0.3.0

Attractor pipeline as a StreamWeave graph
Documentation 
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

//! Tests for `FindStartNode`.

use std::collections::HashMap;
use std::sync::Arc;

use futures::StreamExt;
use streamweave::node::Node;
use tokio_stream::wrappers::ReceiverStream;

use super::FindStartNode;
use super::find_start::find_start_id;
use super::find_start::process_find_start_item;

#[tokio::test]
async fn node_execute_skips_wrong_type() {
  let node = FindStartNode::new("find");
  let (tx, rx) = tokio::sync::mpsc::channel(4);
  tx.send(Arc::new(1_u8) as Arc<dyn std::any::Any + Send + Sync>)
    .await
    .unwrap();
  drop(tx);
  let mut inputs: streamweave::node::InputStreams = HashMap::new();
  inputs.insert(
    "in".to_string(),
    Box::pin(ReceiverStream::new(rx))
      as std::pin::Pin<
        Box<dyn futures::Stream<Item = Arc<dyn std::any::Any + Send + Sync>> + Send>,
      >,
  );
  let mut outputs = node.execute(inputs).await.unwrap();
  let mut out = outputs.remove("out").unwrap();
  let item: Option<Arc<dyn std::any::Any + Send + Sync>> = out.next().await;
  assert!(item.is_none());
}

#[tokio::test]
async fn node_execute_emits_start_id() {
  let dot = r#"digraph G { start [shape=Mdiamond] exit [shape=Msquare] start -> exit }"#;
  let graph = crate::dot_parser::parse_dot(dot).unwrap();
  let node = FindStartNode::new("find");
  let (tx, rx) = tokio::sync::mpsc::channel(4);
  tx.send(Arc::new(graph) as Arc<dyn std::any::Any + Send + Sync>)
    .await
    .unwrap();
  drop(tx);
  let mut inputs: streamweave::node::InputStreams = HashMap::new();
  inputs.insert(
    "in".to_string(),
    Box::pin(ReceiverStream::new(rx))
      as std::pin::Pin<
        Box<dyn futures::Stream<Item = Arc<dyn std::any::Any + Send + Sync>> + Send>,
      >,
  );
  let mut outputs = node.execute(inputs).await.unwrap();
  let mut out = outputs.remove("out").unwrap();
  let item: Option<Arc<dyn std::any::Any + Send + Sync>> = out.next().await;
  assert!(item.is_some());
  let id = item.unwrap().downcast::<String>().unwrap();
  assert_eq!(*id, "start");
}

#[test]
fn node_trait_methods() {
  let mut node = FindStartNode::new("find");
  assert_eq!(node.name(), "find");
  node.set_name("start");
  assert_eq!(node.name(), "start");
  assert!(node.has_input_port("in"));
  assert!(node.has_output_port("out"));
}

#[test]
fn new_creates_node() {
  let n = FindStartNode::new("find");
  assert_eq!(n.name(), "find");
  assert!(n.has_input_port("in"));
  assert!(n.has_output_port("out"));
}

#[test]
fn find_start_id_returns_start_node_id() {
  let dot = r#"digraph G { start [shape=Mdiamond] exit [shape=Msquare] start -> exit }"#;
  let graph = crate::dot_parser::parse_dot(dot).unwrap();
  let id = find_start_id(&graph);
  assert_eq!(id, Some("start".to_string()));
}

#[test]
fn find_start_id_returns_none_when_no_start() {
  let dot = r#"digraph G { a -> b }"#;
  let graph = crate::dot_parser::parse_dot(dot).unwrap();
  let id = find_start_id(&graph);
  assert!(id.is_none());
}

#[test]
fn process_find_start_item_returns_some_for_graph_with_start() {
  let dot = r#"digraph G { start [shape=Mdiamond] exit [shape=Msquare] start -> exit }"#;
  let graph = crate::dot_parser::parse_dot(dot).unwrap();
  let item = Arc::new(graph) as Arc<dyn std::any::Any + Send + Sync>;
  let id = process_find_start_item(item);
  assert_eq!(id, Some("start".to_string()));
}

#[test]
fn process_find_start_item_returns_none_for_wrong_type() {
  let item = Arc::new(42_u32) as Arc<dyn std::any::Any + Send + Sync>;
  let id = process_find_start_item(item);
  assert!(id.is_none());
}