asupersync 0.3.0

Spec-first, cancel-correct, capability-secure async runtime for Rust.
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
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160

//! Transport layer errors.

use std::io;
use thiserror::Error;

/// Errors that can occur when receiving symbols from a stream.
#[derive(Debug, Error)]
pub enum StreamError {
    /// The connection was closed by the peer.
    #[error("Connection closed")]
    Closed,

    /// A stream helper future was polled again after it had already completed.
    #[error("stream future polled after completion")]
    PolledAfterCompletion,

    /// The connection was reset.
    #[error("Connection reset")]
    Reset,

    /// Timed out waiting for a symbol.
    #[error("Timeout waiting for symbol")]
    Timeout,

    /// Authentication failed for a symbol.
    #[error("Authentication failed: {reason}")]
    AuthenticationFailed {
        /// The reason for authentication failure.
        reason: String,
    },

    /// Protocol violation or invalid data.
    #[error("Protocol error: {details}")]
    ProtocolError {
        /// Details about the protocol violation.
        details: String,
    },

    /// Underlying I/O error.
    #[error("I/O error: {source}")]
    Io {
        /// The source I/O error.
        #[from]
        source: io::Error,
    },

    /// Operation was cancelled.
    #[error("Cancelled")]
    Cancelled,
}

/// Errors that can occur when sending symbols to a sink.
#[derive(Debug, Error)]
pub enum SinkError {
    /// The connection was closed.
    #[error("Connection closed")]
    Closed,

    /// A sink helper future was polled again after it had already completed.
    #[error("sink future polled after completion")]
    PolledAfterCompletion,

    /// The internal buffer is full and cannot accept more items.
    #[error("Buffer full")]
    BufferFull,

    /// Failed to send the symbol.
    #[error("Send failed: {reason}")]
    SendFailed {
        /// The reason for the send failure.
        reason: String,
    },

    /// Underlying I/O error.
    #[error("I/O error: {source}")]
    Io {
        /// The source I/O error.
        #[from]
        source: io::Error,
    },

    /// Operation was cancelled.
    #[error("Cancelled")]
    Cancelled,
}

#[cfg(test)]
mod tests {
    use super::*;

    // =========================================================================
    // Pure data-type tests (wave 41 – CyanBarn)
    // =========================================================================

    #[test]
    fn stream_error_debug_display() {
        let closed = StreamError::Closed;
        assert!(format!("{closed:?}").contains("Closed"));
        assert_eq!(format!("{closed}"), "Connection closed");

        let done = StreamError::PolledAfterCompletion;
        assert_eq!(format!("{done}"), "stream future polled after completion");

        let reset = StreamError::Reset;
        assert_eq!(format!("{reset}"), "Connection reset");

        let timeout = StreamError::Timeout;
        assert_eq!(format!("{timeout}"), "Timeout waiting for symbol");

        let cancelled = StreamError::Cancelled;
        assert_eq!(format!("{cancelled}"), "Cancelled");

        let auth = StreamError::AuthenticationFailed {
            reason: "bad token".into(),
        };
        assert!(format!("{auth}").contains("bad token"));

        let proto = StreamError::ProtocolError {
            details: "invalid frame".into(),
        };
        assert!(format!("{proto}").contains("invalid frame"));
    }

    #[test]
    fn stream_error_from_io() {
        let io_err = io::Error::new(io::ErrorKind::BrokenPipe, "pipe broken");
        let stream_err: StreamError = io_err.into();
        assert!(format!("{stream_err}").contains("pipe broken"));
        assert!(matches!(stream_err, StreamError::Io { .. }));
    }

    #[test]
    fn sink_error_debug_display() {
        let closed = SinkError::Closed;
        assert!(format!("{closed:?}").contains("Closed"));
        assert_eq!(format!("{closed}"), "Connection closed");

        let done = SinkError::PolledAfterCompletion;
        assert_eq!(format!("{done}"), "sink future polled after completion");

        let full = SinkError::BufferFull;
        assert_eq!(format!("{full}"), "Buffer full");

        let cancelled = SinkError::Cancelled;
        assert_eq!(format!("{cancelled}"), "Cancelled");

        let send_failed = SinkError::SendFailed {
            reason: "queue overflow".into(),
        };
        assert!(format!("{send_failed}").contains("queue overflow"));
    }

    #[test]
    fn sink_error_from_io() {
        let io_err = io::Error::new(io::ErrorKind::ConnectionRefused, "refused");
        let sink_err: SinkError = io_err.into();
        assert!(format!("{sink_err}").contains("refused"));
        assert!(matches!(sink_err, SinkError::Io { .. }));
    }
}