stan 0.0.16

NATS Streaming client based on official NATS rust client
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

use std::{io, thread, time};
mod utils;

fn next_message_starting_at(
    sc: &stan::Client,
    subject: &str,
    start: stan::SubscriptionStart,
) -> io::Result<String> {
    let sub = sc.subscribe(
        subject,
        stan::SubscriptionConfig {
            start,
            ..Default::default()
        },
    )?;
    utils::next_message(&sub)
}

#[test]
fn test_subscription_start() -> io::Result<()> {
    let server = utils::server()?;
    let nats_url = &format!("localhost:{}", server.port);
    let nc = nats::connect(nats_url)?;
    let subject = "foo";
    let sc = stan::connect(nc, "test-cluster", "rust-client-1")?;
    sc.publish(subject, "hello from rust 1")?;
    let from_time_start = time::SystemTime::now();
    thread::sleep(time::Duration::from_millis(50));
    sc.publish(subject, "hello from rust 2")?;
    thread::sleep(time::Duration::from_millis(50));
    sc.publish(subject, "hello from rust 3")?;
    sc.publish(subject, "hello from rust 4")?;
    sc.publish(subject, "hello from rust 5")?;

    assert_eq!(
        "hello from rust 1",
        next_message_starting_at(&sc, subject, stan::SubscriptionStart::AllAvailable)?,
    );

    assert_eq!(
        "hello from rust 2",
        next_message_starting_at(
            &sc,
            subject,
            stan::SubscriptionStart::FromTimestamp(from_time_start)
        )?
    );

    assert_eq!(
        "hello from rust 3",
        next_message_starting_at(
            &sc,
            subject,
            stan::SubscriptionStart::FromPast(time::Duration::from_millis(50))
        )?,
    );

    assert_eq!(
        "hello from rust 4",
        next_message_starting_at(&sc, subject, stan::SubscriptionStart::FromSequence(4))?,
    );

    assert_eq!(
        "hello from rust 5",
        next_message_starting_at(&sc, subject, stan::SubscriptionStart::LastReceived)?
    );

    let new_only_sub = sc.subscribe(
        subject,
        stan::SubscriptionConfig {
            start: stan::SubscriptionStart::NewOnly,
            ..Default::default()
        },
    )?;

    assert_eq!(
        Err(io::ErrorKind::TimedOut),
        utils::next_message(&new_only_sub).map_err(|e| e.kind())
    );

    sc.publish(subject, "hello from rust 6")?;

    assert_eq!("hello from rust 6", utils::next_message(&new_only_sub)?);

    Ok(())
}