streamweave 0.10.1

Composable, async, stream-first computation in pure Rust
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

//! # Not Node Test Suite

use crate::node::{InputStreams, Node};
use crate::nodes::boolean_logic::not_node::NotNode;
use std::any::Any;
use std::collections::HashMap;
use std::sync::Arc;
use tokio::sync::mpsc;
use tokio_stream::{StreamExt, wrappers::ReceiverStream};

fn create_single_input_stream() -> (mpsc::Sender<Arc<dyn Any + Send + Sync>>, InputStreams) {
  let (_config_tx, config_rx) = mpsc::channel(10);
  let (in_tx, in_rx) = mpsc::channel(10);

  let mut inputs = HashMap::new();
  inputs.insert(
    "configuration".to_string(),
    Box::pin(ReceiverStream::new(config_rx)) as crate::node::InputStream,
  );
  inputs.insert(
    "in".to_string(),
    Box::pin(ReceiverStream::new(in_rx)) as crate::node::InputStream,
  );

  (in_tx, inputs)
}

#[tokio::test]
async fn test_not_node_creation() {
  let node = NotNode::new("test_not".to_string());
  assert_eq!(node.name(), "test_not");
  assert!(node.has_input_port("configuration"));
  assert!(node.has_input_port("in"));
  assert!(node.has_output_port("out"));
  assert!(node.has_output_port("error"));
}

#[tokio::test]
async fn test_not_node_inverts_true() {
  let node = NotNode::new("test_not".to_string());
  let (in_tx, inputs) = create_single_input_stream();

  let outputs_future = node.execute(inputs);
  let mut outputs = outputs_future.await.unwrap();

  let _ = in_tx
    .send(Arc::new(true) as Arc<dyn Any + Send + Sync>)
    .await;
  tokio::time::sleep(tokio::time::Duration::from_millis(10)).await;

  let out_stream = outputs.remove("out").unwrap();
  let mut results = Vec::new();
  let mut stream = out_stream;

  tokio::select! {
    result = stream.next() => {
      if let Some(item) = result
        && let Ok(arc_bool) = item.downcast::<bool>() {
          results.push(*arc_bool);
        }
    }
    _ = tokio::time::sleep(tokio::time::Duration::from_millis(100)) => {}
  }

  assert_eq!(results.len(), 1);
  assert!(!results[0]); // !true = false
}

#[tokio::test]
async fn test_not_node_inverts_false() {
  let node = NotNode::new("test_not".to_string());
  let (in_tx, inputs) = create_single_input_stream();

  let outputs_future = node.execute(inputs);
  let mut outputs = outputs_future.await.unwrap();

  let _ = in_tx
    .send(Arc::new(false) as Arc<dyn Any + Send + Sync>)
    .await;
  tokio::time::sleep(tokio::time::Duration::from_millis(10)).await;

  let out_stream = outputs.remove("out").unwrap();
  let mut results = Vec::new();
  let mut stream = out_stream;

  tokio::select! {
    result = stream.next() => {
      if let Some(item) = result
        && let Ok(arc_bool) = item.downcast::<bool>() {
          results.push(*arc_bool);
        }
    }
    _ = tokio::time::sleep(tokio::time::Duration::from_millis(100)) => {}
  }

  assert_eq!(results.len(), 1);
  assert!(results[0]); // !false = true
}