rants 0.6.0

An async NATS client library.
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

mod common;

use common::NatsServer;
use rants::{error::Error, Client, Subject};

async fn make_subscription(client: Client, subject: &Subject) {
    let mut subscription = client.subscribe(subject, 1).await.unwrap().1;
    let client_copy = Client::clone(&client);
    tokio::spawn(async move {
        // Wait for the request
        let request = subscription.recv().await.unwrap();
        let reply_to = request.reply_to().unwrap().clone();
        let request = String::from_utf8(request.into_payload()).unwrap();
        assert_eq!(&request, "the request");
        // Send the reply
        let client = client_copy;
        client.publish(&reply_to, b"the reply").await.unwrap();
    });
}

#[tokio::test(flavor = "multi_thread")]
async fn request() {
    common::init();
    let _nats_server = NatsServer::new(&["--auth=abc123"]).await;

    let address = "127.0.0.1".parse().unwrap();
    let client = Client::new(vec![address]);
    client
        .connect_mut()
        .await
        .echo(true)
        .token(String::from("abc123"));
    client.connect().await;

    // Spawn a few subscription tasks to send the reply
    let subject1 = "test_subject1".parse::<Subject>().unwrap();
    make_subscription(Client::clone(&client), &subject1).await;
    let subject2 = "test_subject2".parse::<Subject>().unwrap();
    make_subscription(Client::clone(&client), &subject2).await;

    // Make a first request
    let reply = client.request(&subject1, b"the request").await.unwrap();
    let reply = String::from_utf8(reply.into_payload()).unwrap();
    assert_eq!(&reply, "the reply");

    // Make a second request
    let reply = client.request(&subject2, b"the request").await.unwrap();
    let reply = String::from_utf8(reply.into_payload()).unwrap();
    assert_eq!(&reply, "the reply");

    // Make a third request (via timeout interface to show it's wired correctly)
    let subject3 = "test_subject3".parse::<Subject>().unwrap();
    make_subscription(Client::clone(&client), &subject3).await;
    let reply = client
        .request_with_timeout(
            &subject3,
            b"the request",
            std::time::Duration::from_millis(1000),
        )
        .await
        .unwrap();
    let reply = String::from_utf8(reply.into_payload()).unwrap();
    assert_eq!(&reply, "the reply");

    // Make a fourth request (that will timeout)
    let subject4 = "test_subject4".parse::<Subject>().unwrap();
    let err = client
        .request_with_timeout(
            &subject4,
            b"the request",
            std::time::Duration::from_millis(10),
        )
        .await
        .unwrap_err();

    // Ensure we get a timeout error. Fail otherwise.
    match err {
        Error::Timeout(_) => {}
        _ => assert!(false),
    };

    client.disconnect().await;
}