fiddler 4.9.1

Data Stream processor written in 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
99
100
101
102
103
104
105
106

#![allow(unused_crate_dependencies)]
#![allow(missing_docs)]
#[cfg(feature = "aws")]
use aws_sdk_sqs as sqs;
#[cfg(feature = "aws")]
use aws_sdk_sqs::config::BehaviorVersion;
#[cfg(feature = "aws")]
use aws_sdk_sqs::config::Region;
#[cfg(feature = "aws")]
use fiddler::Runtime;
#[cfg(feature = "aws")]
use testcontainers::{runners::AsyncRunner, ImageExt};
#[cfg(feature = "aws")]
use testcontainers_modules::localstack::LocalStack;

#[cfg(feature = "aws")]
#[cfg_attr(feature = "aws", tokio::test)]
async fn fiddler_aws_sqs_test() {
    let request = LocalStack::default().with_env_var("SERVICES", "sqs");
    let container = request.start().await.unwrap();

    let host_ip = container.get_host().await.unwrap();
    let host_port = container.get_host_port_ipv4(4566).await.unwrap();
    let endpoint_url = format!("http://{host_ip}:{host_port}");
    let creds = sqs::config::Credentials::new("fake", "fake", None, None, "test");
    std::env::set_var("AWS_ACCESS_KEY_ID", "fake");
    std::env::set_var("AWS_SECRET_ACCESS_KEY", "fake");
    std::env::set_var("AWS_DEFAULT_REGION", "us-east-1");

    let config = sqs::config::Builder::default()
        .behavior_version(BehaviorVersion::v2025_01_17())
        .region(Region::new("us-east-1"))
        .credentials_provider(creds)
        .endpoint_url(&endpoint_url)
        .build();

    let client = sqs::Client::from_conf(config);

    let in_queue = client
        .create_queue()
        .queue_name("in_queue")
        .send()
        .await
        .unwrap();

    let in_queue_url = in_queue.queue_url().unwrap();

    let _ = client
        .send_message()
        .queue_url(in_queue_url)
        .message_body("Testing Message to send to SQS")
        .send()
        .await
        .unwrap();

    let out_queue = client
        .create_queue()
        .queue_name("out_queue")
        .send()
        .await
        .unwrap();

    let out_queue_url = out_queue.queue_url().unwrap();

    let config = format!(
        "input:
  aws_sqs:
    queue_url: {}
    endpoint_url: {}
    region: us-east-1
    credentials: 
      access_key_id: fake
      secret_access_key: fake
num_threads: 1
processors: []
output:
  aws_sqs:
    queue_url: {}
    endpoint_url: {}
    region: us-east-1
    credentials: 
      access_key_id: fake
      secret_access_key: fake",
        in_queue_url, endpoint_url, out_queue_url, endpoint_url
    );

    let mut env = Runtime::from_config(&config).await.unwrap();
    env.set_timeout(Some(tokio::time::Duration::from_secs(5)))
        .unwrap();
    env.run().await.unwrap();

    let result = client
        .receive_message()
        .queue_url(out_queue_url)
        .send()
        .await
        .unwrap();

    assert_eq!(result.messages.is_some(), true);
    let messages = result.messages();
    assert_eq!(messages.len(), 1);
    assert_eq!(
        messages[0].body().unwrap(),
        "Testing Message to send to SQS"
    );
}