ruststream_zeromq/
message.rs

1//! [`ZmqMessage`]: a delivered message.
2
3use std::future::{Future, ready};
4
5use bytes::Bytes;
6use ruststream::{AckError, HeaderMap, IncomingMessage};
7
8#[cfg(feature = "testing")]
9use crate::in_process::Release;
10
11/// A message delivered by one of the transport's subscribers.
12///
13/// Delivery is at most once and there is no durability, so acknowledgement is reported as
14/// [`AckError::Unsupported`] rather than emulated.
15pub struct ZmqMessage {
16    pub(crate) name: String,
17    pub(crate) headers: HeaderMap,
18    pub(crate) payload: Bytes,
19    /// What the in-process transport hands the test harness back once this delivery is gone.
20    #[cfg(feature = "testing")]
21    release: Release,
22}
23
24// The in-process mode adds nothing to a production delivery: without the `testing` feature the
25// message is its three frames and nothing else.
26#[cfg(not(feature = "testing"))]
27const _: () = assert!(size_of::<ZmqMessage>() == size_of::<(String, HeaderMap, Bytes)>());
28
29impl std::fmt::Debug for ZmqMessage {
30    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
31        f.debug_struct("ZmqMessage")
32            .field("name", &self.name)
33            .field("payload_len", &self.payload.len())
34            .finish_non_exhaustive()
35    }
36}
37
38impl ZmqMessage {
39    /// A delivery made of the frames a socket received.
40    pub(crate) fn new(name: String, headers: HeaderMap, payload: Bytes) -> Self {
41        Self {
42            name,
43            headers,
44            payload,
45            #[cfg(feature = "testing")]
46            release: Release::default(),
47        }
48    }
49
50    /// Hands the harness's count of this delivery to the delivery itself, which gives it back
51    /// once it is settled or dropped.
52    #[cfg(feature = "testing")]
53    pub(crate) fn counted(mut self, release: Release) -> Self {
54        self.release = release;
55        self
56    }
57
58    /// The name frame this message carried.
59    #[must_use]
60    pub fn name(&self) -> &str {
61        &self.name
62    }
63}
64
65impl IncomingMessage for ZmqMessage {
66    fn payload(&self) -> &[u8] {
67        &self.payload
68    }
69
70    fn headers(&self) -> &HeaderMap {
71        &self.headers
72    }
73
74    fn ack(self) -> impl Future<Output = Result<(), AckError>> {
75        ready(Err(AckError::Unsupported))
76    }
77
78    fn nack(self, _requeue: bool) -> impl Future<Output = Result<(), AckError>> {
79        ready(Err(AckError::Unsupported))
80    }
81}
ruststream_zeromq/message.rs

ruststream_zeromq/
message.rs
