dex-sdk 1.38.0

Strongly typed Rust SDK for Dex
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
110
111
112
113
114
115
116
117
118
119

// Portions of this file are derived from indeedeng/iwf-java-sdk.
// Those portions are licensed under the Apache License, Version 2.0.
// See LICENSES/Apache-2.0.txt and LEGACY_NOTICES.md.
//
// Modifications Copyright (c) 2026 Super Durable, Inc.
//
// Modifications are licensed under the Sustainable Use License 1.0.
// Third-Party Materials remain under the Apache License, Version 2.0.
// See LICENSE and LEGACY_NOTICES.md.

use std::time::{Duration, SystemTime};

use dex_sdk::Registry;

use crate::stream_workflow::{PROGRESS, StreamTestWorkflow};
use crate::support::{DexDevTestEnvironment, flow_id};

#[test]
#[ignore = "requires dexcli dev"]
fn test_stream_round_trip() {
    let workflow = StreamTestWorkflow::new();
    let environment = DexDevTestEnvironment::start(Registry::new().register(workflow.clone()));
    let flow_id = flow_id("stream");
    environment
        .client
        .start_flow(&workflow, &flow_id, ())
        .expect("start Stream Flow");
    environment
        .client
        .wait_for_flow_with_timeout(&flow_id, Duration::from_secs(30))
        .expect("complete Stream Flow");

    environment
        .client
        .write_stream(
            &flow_id,
            &PROGRESS,
            "client#source",
            "client-progress".to_string(),
        )
        .expect("write client Stream message");
    environment
        .client
        .write_stream(
            &flow_id,
            &PROGRESS,
            "client#source",
            "client-progress-again".to_string(),
        )
        .expect("append repeated-source client Stream message");

    let mut resume_token = String::new();
    let mut step_source = None;
    for expected in [
        "wait-progress-1",
        "wait-progress-2",
        "execute-progress-1",
        "execute-progress-2",
    ] {
        let message = environment
            .client
            .read_stream_with_timeout(&flow_id, &PROGRESS, &resume_token, Duration::from_secs(30))
            .expect("read Step Stream message");
        assert_eq!(expected, message.value);
        assert!(!message.resume_token.is_empty());
        assert!(message.created_time > SystemTime::UNIX_EPOCH);
        assert!(message.source.starts_with('#'));
        assert!(!message.source[1..].is_empty());
        if let Some(source) = step_source.as_ref() {
            assert_eq!(source, &message.source);
        } else {
            step_source = Some(message.source.clone());
        }
        resume_token = message.resume_token;
    }

    for expected in ["client-progress", "client-progress-again"] {
        let message = environment
            .client
            .read_stream_with_timeout(&flow_id, &PROGRESS, &resume_token, Duration::from_secs(30))
            .expect("read client Stream message");
        assert_eq!(expected, message.value);
        assert_ne!(resume_token, message.resume_token);
        assert!(message.created_time > SystemTime::UNIX_EPOCH);
        assert_eq!("client#source", message.source);
        resume_token = message.resume_token;
    }

    let latest = environment
        .client
        .list_stream_messages(&flow_id, &PROGRESS, 3, "")
        .expect("list latest Stream messages");
    assert_eq!(
        vec![
            "client-progress-again",
            "client-progress",
            "execute-progress-2",
        ],
        latest
            .messages
            .iter()
            .map(|message| message.value.as_str())
            .collect::<Vec<_>>()
    );
    assert!(!latest.next_page_token.is_empty());
    let older = environment
        .client
        .list_stream_messages(&flow_id, &PROGRESS, 3, &latest.next_page_token)
        .expect("list older Stream messages");
    assert_eq!(
        vec!["execute-progress-1", "wait-progress-2", "wait-progress-1"],
        older
            .messages
            .iter()
            .map(|message| message.value.as_str())
            .collect::<Vec<_>>()
    );
    assert!(older.next_page_token.is_empty());
}