mq-bridge 0.2.21

An asynchronous message bridging library connecting Kafka, MQTT, AMQP, NATS, MongoDB, HTTP, and more.
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

#![allow(unused_imports, dead_code)]

use mq_bridge::models::{Endpoint, Route};
use mq_bridge::{CanonicalMessage, Handled, Publisher, Sent};
use std::time::Duration;

#[tokio::test(flavor = "multi_thread")]
async fn test_memory_request_reply_logic() {
    // This test verifies the Request -> Route -> Handler -> Response Endpoint flow.
    let in_topic = "req_rep_in";

    // 1. Define a route with a handler that produces a reply.
    let handler = |mut msg: CanonicalMessage| async move {
        let payload = msg.get_payload_str();
        msg.set_payload_str(format!("reply_to_{}", payload));
        Ok(Handled::Publish(msg))
    };

    let route = Route::new(
        Endpoint::new_memory(in_topic, 100),
        Endpoint::new_response(),
    )
    .with_handler(handler);

    route.deploy("logic_req_rep").await.unwrap();

    // 2. Create a publisher with request_reply mode enabled.
    let mut config = mq_bridge::models::MemoryConfig::new(in_topic, Some(100));
    config.request_reply = true;
    config.request_timeout_ms = Some(2000);

    let publisher = Publisher::new(Endpoint {
        endpoint_type: mq_bridge::models::EndpointType::Memory(config),
        ..Default::default()
    })
    .await
    .unwrap();

    // 3. Send request and verify response.
    let result = publisher.send("hello".into()).await.unwrap();

    if let Sent::Response(resp) = result {
        assert_eq!(resp.get_payload_str(), "reply_to_hello");
    } else {
        panic!("Expected Sent::Response, got {:?}", result);
    }

    Route::stop("logic_req_rep").await;
}

#[cfg(feature = "http")]
#[tokio::test(flavor = "multi_thread")]
async fn test_http_request_reply_pattern() {
    // Verifies Request-Reply over a real HTTP boundary.
    use std::net::TcpListener;
    // Bind to an ephemeral port
    let listener = TcpListener::bind("127.0.0.1:0").unwrap();
    let port = listener.local_addr().unwrap().port();
    drop(listener); // Release the port for the HTTP server
    let addr = format!("127.0.0.1:{}", port);

    let http_in = Endpoint {
        endpoint_type: mq_bridge::models::EndpointType::Http(mq_bridge::models::HttpConfig {
            url: addr.clone(),
            ..Default::default()
        }),
        ..Default::default()
    };

    let handler = |mut msg: CanonicalMessage| async move {
        msg.set_payload_str("http_pong");
        Ok(Handled::Publish(msg))
    };

    let route = Route::new(http_in, Endpoint::new_response()).with_handler(handler);
    route.deploy("http_logic").await.unwrap();

    // Wait for HTTP server readiness (retry loop)
    let client = reqwest::Client::new();
    let mut ready = false;
    for _ in 0..20 {
        let resp = client
            .post(format!("http://{}", addr))
            .body("ping")
            .send()
            .await;
        if let Ok(res) = &resp {
            if res.status() == reqwest::StatusCode::OK {
                ready = true;
                break;
            }
        }
        tokio::time::sleep(std::time::Duration::from_millis(50)).await;
    }
    assert!(ready, "HTTP server did not become ready in time");

    // Final request for assertion
    let res = client
        .post(format!("http://{}", addr))
        .body("ping")
        .send()
        .await
        .unwrap();

    assert_eq!(res.status(), reqwest::StatusCode::OK);
    assert_eq!(res.text().await.unwrap(), "http_pong");

    Route::stop("http_logic").await;
}