topmesys 0.2.1

an embeddable topic-based messaging system
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

use std::{fmt::Display, sync::Arc};

use crate::{
    EventMessage, EventTopic, SubscriptionInfo, TransportHandle,
    transport::{DeliveryOutcome, Settler},
    type_states::Pattern,
};

/// A message handed to an [EventConsumer](crate::EventConsumer) on one of its subscriptions.
#[derive(Debug)]
pub struct Delivery {
    pub(crate) message: Arc<EventMessage>,
    pub(crate) subscription: Arc<SubscriptionInfo>,
    pub(crate) attempt: u32,
    pub(crate) settler: Option<Arc<Settler>>,
}

impl Delivery {
    /// Creates a delivery on the given pattern outside of a broker, e.g. to test an
    /// [EventConsumer](crate::EventConsumer). Its [consumer](SubscriptionInfo::consumer) is empty and
    /// it never settles the message.
    pub fn new(message: impl Into<Arc<EventMessage>>, pattern: EventTopic<Pattern>) -> Self {
        Self {
            message: message.into(),
            subscription: Arc::new(SubscriptionInfo {
                pattern,
                consumer: "",
            }),
            attempt: 1,
            settler: None,
        }
    }

    /// Self-consuming attempt setter
    pub fn with_attempt(mut self, attempt: u32) -> Self {
        self.attempt = attempt;
        self
    }

    pub fn message(&self) -> &Arc<EventMessage> {
        &self.message
    }

    pub fn subscription(&self) -> &SubscriptionInfo {
        &self.subscription
    }

    /// The current attempt at handling the message, starting at `1`.
    pub fn attempt(&self) -> u32 {
        self.attempt
    }

    /// The message's transport handle, if it carries one of type `T`.
    pub fn transport<T: TransportHandle>(&self) -> Option<&T> {
        self.message.transport()
    }

    /// Records the delivery's outcome, settling the message if no other delivery of it is open.
    pub(crate) async fn finish(mut self, outcome: DeliveryOutcome) {
        let Some(settler) = self.settler.take() else {
            return;
        };
        settler.record(&self.subscription, outcome);
        if let Some(settler) = Arc::into_inner(settler) {
            settler.settle().await;
        }
    }
}

impl Drop for Delivery {
    /// A delivery still holding its settler was never finished, e.g. because its worker was aborted.
    /// It is recorded as aborted, so the message still settles and its transport can redeliver it.
    fn drop(&mut self) {
        let Some(settler) = self.settler.take() else {
            return;
        };
        settler.record(&self.subscription, DeliveryOutcome::Aborted);
        if let Some(settler) = Arc::into_inner(settler) {
            match tokio::runtime::Handle::try_current() {
                Ok(runtime) => {
                    runtime.spawn(settler.settle());
                }
                Err(_) => tracing::error!(
                    "Unable to settle aborted event on `{}`: no tokio runtime available",
                    self.message.topic()
                ),
            }
        }
    }
}

/// The error returned by [EventConsumer::handle_event](crate::EventConsumer::handle_event). Any
/// error converts into a transient `HandlerError` with `?`, which is retried according to the
/// subscription's [RetryPolicy](crate::RetryPolicy). Errors retrying won't fix should be returned as
/// [permanent](HandlerError::permanent), which dead-letters the message right away. Like
/// [anyhow::Error], `HandlerError` doesn't implement [std::error::Error] itself, as that would rule
/// out the conversion.
#[derive(Debug)]
pub struct HandlerError {
    error: anyhow::Error,
    permanent: bool,
}

impl HandlerError {
    /// An error worth retrying, e.g. a timeout. Same as converting the error with `?`.
    pub fn transient(error: impl Into<anyhow::Error>) -> Self {
        Self {
            error: error.into(),
            permanent: false,
        }
    }

    /// An error retrying won't fix, e.g. a malformed payload.
    pub fn permanent(error: impl Into<anyhow::Error>) -> Self {
        Self {
            error: error.into(),
            permanent: true,
        }
    }

    pub fn is_permanent(&self) -> bool {
        self.permanent
    }

    pub fn error(&self) -> &anyhow::Error {
        &self.error
    }

    pub fn into_error(self) -> anyhow::Error {
        self.error
    }
}

impl<E> From<E> for HandlerError
where
    E: Into<anyhow::Error>,
{
    fn from(error: E) -> Self {
        Self::transient(error)
    }
}

impl Display for HandlerError {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        Display::fmt(&self.error, f)
    }
}