wireframe 0.3.0

Simplify building servers and clients for custom binary protocols.
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

//! Tracing tests for streaming client operations.
//!
//! These tests verify that `call_streaming` and the response stream emit the
//! expected tracing spans and events. They are separated from the core tracing
//! tests because they depend on the streaming test infrastructure.

use futures::StreamExt;
use rstest::rstest;
use tracing_test::traced_test;

use super::{
    streaming_infra::{
        CorrelationId,
        MessageId,
        Payload,
        TestStreamEnvelope,
        setup_streaming_test,
        spawn_test_server,
    },
    tracing::span_assertion,
};
use crate::client::{TracingConfig, WireframeClient};

#[rstest]
#[traced_test]
#[tokio::test]
async fn call_streaming_emits_span_with_correlation_id() {
    let cid = CorrelationId::new(42);
    let frames = vec![
        TestStreamEnvelope::data(MessageId::new(1), cid, Payload::new(vec![10])),
        TestStreamEnvelope::terminator(cid),
    ];

    let server = spawn_test_server(frames, false).await.expect("server");
    let mut client = WireframeClient::builder()
        .tracing_config(TracingConfig::default().with_streaming_timing(true))
        .connect(server.addr)
        .await
        .expect("connect");

    let request = TestStreamEnvelope::data(MessageId::new(99), cid, Payload::new(vec![]));
    let mut stream = client
        .call_streaming::<TestStreamEnvelope>(request)
        .await
        .expect("call_streaming");

    // Consume the stream — frame events carry the call_streaming span context.
    while let Some(result) = stream.next().await {
        let _frame = result.expect("data frame");
    }

    // With streaming timing enabled, the elapsed_us event fires within
    // the client.call_streaming span, making the span name visible.
    logs_assert(span_assertion("client.call_streaming", &["correlation_id"]));
}

#[rstest]
#[traced_test]
#[tokio::test]
async fn response_stream_emits_frame_events() {
    let cid = CorrelationId::new(42);
    let frames = vec![
        TestStreamEnvelope::data(MessageId::new(1), cid, Payload::new(vec![10])),
        TestStreamEnvelope::data(MessageId::new(2), cid, Payload::new(vec![20])),
        TestStreamEnvelope::data(MessageId::new(3), cid, Payload::new(vec![30])),
        TestStreamEnvelope::terminator(cid),
    ];

    let (mut client, _server) = setup_streaming_test(frames).await.expect("setup");

    let request = TestStreamEnvelope::data(MessageId::new(99), cid, Payload::new(vec![]));
    let mut stream = client
        .call_streaming::<TestStreamEnvelope>(request)
        .await
        .expect("call_streaming");

    while let Some(result) = stream.next().await {
        let _frame = result.expect("data frame");
    }

    // Only data frames increment the counter; the terminator does not.
    assert_eq!(
        stream.frame_count(),
        3,
        "frame_count should only include successfully decoded data frames"
    );

    logs_assert(|lines: &[&str]| {
        let frame_events = lines
            .iter()
            .filter(|line| line.contains("stream frame received"))
            .count();
        if frame_events < 3 {
            return Err(format!(
                "expected at least 3 'stream frame received' events, found {frame_events}"
            ));
        }
        lines
            .iter()
            .find(|line| line.contains("stream terminated"))
            .map(|_| ())
            .ok_or_else(|| "'stream terminated' event not found".to_string())
    });
}