aws_smithy_types/
event_stream.rs

1/*
2 * Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
3 * SPDX-License-Identifier: Apache-2.0
4 */
5
6//! Types relevant to event stream serialization/deserialization
7
8use crate::config_bag::{Storable, StoreReplace};
9use crate::str_bytes::StrBytes;
10use bytes::Bytes;
11use std::any::Any;
12use std::fmt;
13use std::sync::{mpsc, Mutex};
14
15mod value {
16    use crate::str_bytes::StrBytes;
17    use crate::DateTime;
18    use bytes::Bytes;
19
20    /// Event Stream frame header value.
21    #[non_exhaustive]
22    #[derive(Clone, Debug, PartialEq)]
23    pub enum HeaderValue {
24        /// Represents a boolean value.
25        Bool(bool),
26        /// Represents a byte value.
27        Byte(i8),
28        /// Represents an int16 value.
29        Int16(i16),
30        /// Represents an int32 value.
31        Int32(i32),
32        /// Represents an int64 value.
33        Int64(i64),
34        /// Represents a byte array value.
35        ByteArray(Bytes),
36        /// Represents a string value.
37        String(StrBytes),
38        /// Represents a timestamp value.
39        Timestamp(DateTime),
40        /// Represents a uuid value.
41        Uuid(u128),
42    }
43
44    impl HeaderValue {
45        /// If the `HeaderValue` is a `Bool`, returns the associated `bool`. Returns `Err` otherwise.
46        pub fn as_bool(&self) -> Result<bool, &Self> {
47            match self {
48                HeaderValue::Bool(value) => Ok(*value),
49                _ => Err(self),
50            }
51        }
52
53        /// If the `HeaderValue` is a `Byte`, returns the associated `i8`. Returns `Err` otherwise.
54        pub fn as_byte(&self) -> Result<i8, &Self> {
55            match self {
56                HeaderValue::Byte(value) => Ok(*value),
57                _ => Err(self),
58            }
59        }
60
61        /// If the `HeaderValue` is an `Int16`, returns the associated `i16`. Returns `Err` otherwise.
62        pub fn as_int16(&self) -> Result<i16, &Self> {
63            match self {
64                HeaderValue::Int16(value) => Ok(*value),
65                _ => Err(self),
66            }
67        }
68
69        /// If the `HeaderValue` is an `Int32`, returns the associated `i32`. Returns `Err` otherwise.
70        pub fn as_int32(&self) -> Result<i32, &Self> {
71            match self {
72                HeaderValue::Int32(value) => Ok(*value),
73                _ => Err(self),
74            }
75        }
76
77        /// If the `HeaderValue` is an `Int64`, returns the associated `i64`. Returns `Err` otherwise.
78        pub fn as_int64(&self) -> Result<i64, &Self> {
79            match self {
80                HeaderValue::Int64(value) => Ok(*value),
81                _ => Err(self),
82            }
83        }
84
85        /// If the `HeaderValue` is a `ByteArray`, returns the associated [`Bytes`]. Returns `Err` otherwise.
86        pub fn as_byte_array(&self) -> Result<&Bytes, &Self> {
87            match self {
88                HeaderValue::ByteArray(value) => Ok(value),
89                _ => Err(self),
90            }
91        }
92
93        /// If the `HeaderValue` is a `String`, returns the associated [`StrBytes`]. Returns `Err` otherwise.
94        pub fn as_string(&self) -> Result<&StrBytes, &Self> {
95            match self {
96                HeaderValue::String(value) => Ok(value),
97                _ => Err(self),
98            }
99        }
100
101        /// If the `HeaderValue` is a `Timestamp`, returns the associated [`DateTime`]. Returns `Err` otherwise.
102        pub fn as_timestamp(&self) -> Result<DateTime, &Self> {
103            match self {
104                HeaderValue::Timestamp(value) => Ok(*value),
105                _ => Err(self),
106            }
107        }
108
109        /// If the `HeaderValue` is a `Uuid`, returns the associated `u128`. Returns `Err` otherwise.
110        pub fn as_uuid(&self) -> Result<u128, &Self> {
111            match self {
112                HeaderValue::Uuid(value) => Ok(*value),
113                _ => Err(self),
114            }
115        }
116    }
117}
118
119pub use value::HeaderValue;
120
121/// Event Stream header.
122#[non_exhaustive]
123#[derive(Clone, Debug, PartialEq)]
124pub struct Header {
125    name: StrBytes,
126    value: HeaderValue,
127}
128
129impl Header {
130    /// Creates a new header with the given `name` and `value`.
131    pub fn new(name: impl Into<StrBytes>, value: impl Into<HeaderValue>) -> Header {
132        Header {
133            name: name.into(),
134            value: value.into(),
135        }
136    }
137
138    /// Returns the header name.
139    pub fn name(&self) -> &StrBytes {
140        &self.name
141    }
142
143    /// Returns the header value.
144    pub fn value(&self) -> &HeaderValue {
145        &self.value
146    }
147}
148
149/// Event Stream message.
150#[non_exhaustive]
151#[derive(Clone, Debug, PartialEq)]
152pub struct Message {
153    headers: Vec<Header>,
154    payload: Bytes,
155}
156
157impl Message {
158    /// Creates a new message with the given `payload`. Headers can be added later.
159    pub fn new(payload: impl Into<Bytes>) -> Message {
160        Message {
161            headers: Vec::new(),
162            payload: payload.into(),
163        }
164    }
165
166    /// Creates a message with the given `headers` and `payload`.
167    pub fn new_from_parts(headers: Vec<Header>, payload: impl Into<Bytes>) -> Self {
168        Self {
169            headers,
170            payload: payload.into(),
171        }
172    }
173
174    /// Adds a header to the message.
175    pub fn add_header(mut self, header: Header) -> Self {
176        self.headers.push(header);
177        self
178    }
179
180    /// Returns all headers.
181    pub fn headers(&self) -> &[Header] {
182        &self.headers
183    }
184
185    /// Returns the payload bytes.
186    pub fn payload(&self) -> &Bytes {
187        &self.payload
188    }
189}
190
191/// Raw message from an event stream receiver when a response error is encountered.
192#[derive(Debug)]
193#[non_exhaustive]
194pub enum RawMessage {
195    /// Message was decoded into a valid frame, but failed to unmarshall into a modeled type.
196    Decoded(Message),
197    /// Message failed to be decoded into a valid frame. The raw bytes may not be available in the
198    /// case where decoding consumed the buffer.
199    Invalid(Option<Bytes>),
200}
201
202impl RawMessage {
203    /// Creates a `RawMessage` for failure to decode a message into a valid frame.
204    pub fn invalid(bytes: Option<Bytes>) -> Self {
205        Self::Invalid(bytes)
206    }
207}
208
209/// Error returned when sending a deferred signer fails.
210#[derive(Debug)]
211pub struct DeferredSignerSendError {
212    kind: DeferredSignerSendErrorKind,
213}
214
215#[derive(Debug)]
216enum DeferredSignerSendErrorKind {
217    Closed,
218}
219
220impl fmt::Display for DeferredSignerSendError {
221    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
222        match self.kind {
223            DeferredSignerSendErrorKind::Closed => f.write_str("receiver was dropped"),
224        }
225    }
226}
227
228impl std::error::Error for DeferredSignerSendError {}
229
230/// Error returned when receiving a deferred signer fails.
231#[derive(Debug)]
232pub struct DeferredSignerRecvError {
233    kind: DeferredSignerRecvErrorKind,
234}
235
236#[derive(Debug)]
237enum DeferredSignerRecvErrorKind {
238    NoSigner,
239}
240
241impl fmt::Display for DeferredSignerRecvError {
242    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
243        match self.kind {
244            DeferredSignerRecvErrorKind::NoSigner => f.write_str("no signer was available"),
245        }
246    }
247}
248
249impl std::error::Error for DeferredSignerRecvError {}
250
251/// Receiver half of the deferred signer channel.
252///
253/// Held internally by a deferred signer implementation to receive the concrete
254/// signer once it becomes available after HTTP request signing.
255#[derive(Debug)]
256pub struct DeferredSignerReceiver {
257    rx: Mutex<Option<mpsc::Receiver<Box<dyn Any + Send + Sync>>>>,
258}
259
260impl DeferredSignerReceiver {
261    /// Receives the value sent by the corresponding [`DeferredSignerSender`].
262    ///
263    /// This consumes the receiver's channel on first call. Subsequent calls will
264    /// return an error.
265    pub fn recv<T: Send + Sync + 'static>(&self) -> Result<T, DeferredSignerRecvError> {
266        let mut rx = self.rx.lock().unwrap();
267        rx.take()
268            .and_then(|r| r.try_recv().ok())
269            .and_then(|any| any.downcast::<T>().ok())
270            .map(|b| *b)
271            .ok_or(DeferredSignerRecvError {
272                kind: DeferredSignerRecvErrorKind::NoSigner,
273            })
274    }
275}
276
277/// Sender for wiring up an event stream message signer after HTTP request signing.
278///
279/// During serialization, a [`DeferredSignerSender`] is placed in the config bag.
280/// After HTTP signing completes, the auth scheme retrieves it and sends the
281/// concrete signer implementation through the channel.
282#[derive(Debug)]
283pub struct DeferredSignerSender {
284    tx: Mutex<mpsc::Sender<Box<dyn Any + Send + Sync>>>,
285}
286
287impl DeferredSignerSender {
288    /// Creates a new sender/receiver pair.
289    pub fn new() -> (DeferredSignerReceiver, Self) {
290        let (tx, rx) = mpsc::channel();
291        (
292            DeferredSignerReceiver {
293                rx: Mutex::new(Some(rx)),
294            },
295            Self { tx: Mutex::new(tx) },
296        )
297    }
298
299    /// Sends a value through the channel.
300    pub fn send<T: Send + Sync + 'static>(&self, value: T) -> Result<(), DeferredSignerSendError> {
301        self.tx
302            .lock()
303            .unwrap()
304            .send(Box::new(value))
305            .map_err(|_| DeferredSignerSendError {
306                kind: DeferredSignerSendErrorKind::Closed,
307            })
308    }
309}
310
311impl Storable for DeferredSignerSender {
312    type Storer = StoreReplace<Self>;
313}
314
315/// An error that occurs when signing an Event Stream message.
316pub type SignMessageError = Box<dyn std::error::Error + Send + Sync + 'static>;
317
318/// Signs an Event Stream message.
319pub trait SignMessage: fmt::Debug {
320    /// Signs a message, returning the signed version.
321    fn sign(&mut self, message: Message) -> Result<Message, SignMessageError>;
322
323    /// SigV4 requires an empty last signed message to be sent.
324    /// Other protocols do not require one.
325    /// Return `Some(_)` to send a signed last empty message, before completing the stream.
326    /// Return `None` to not send one and terminate the stream immediately.
327    fn sign_empty(&mut self) -> Option<Result<Message, SignMessageError>>;
328}
aws_smithy_types/event_stream.rs

aws_smithy_types/
event_stream.rs
