blueprint_tangle_extra/extract/
block_events.rs

1use blueprint_core::__composite_rejection as composite_rejection;
2use blueprint_core::__define_rejection as define_rejection;
3use blueprint_core::{FromJobCallParts, job::call::Parts as JobCallParts};
4use tangle_subxt::subxt::events::Events;
5use tangle_subxt::subxt::events::StaticEvent;
6
7use crate::producer::TangleConfig;
8
9/// Extracts all the events that happened in the current block.
10#[derive(Debug, Clone)]
11pub struct BlockEvents(pub Events<TangleConfig>);
12
13blueprint_core::__impl_deref!(BlockEvents: Events<TangleConfig>);
14blueprint_core::__impl_from!(Events<TangleConfig>, BlockEvents);
15
16define_rejection! {
17  #[body = "No events found in the extensions. Did you forget to add the `AddBlockEventsLayer`?"]
18  /// A Rejection type for [`BlockEvents`] when it is missing in the Extensions.
19  /// This usually happens when the `AddBlockEventsLayer` is not added to the Router.
20  pub struct MissingBlockEvents;
21}
22
23impl TryFrom<&mut JobCallParts> for BlockEvents {
24    type Error = MissingBlockEvents;
25
26    fn try_from(parts: &mut JobCallParts) -> Result<Self, Self::Error> {
27        let events = parts
28            .extensions
29            .get::<Events<TangleConfig>>()
30            .ok_or(MissingBlockEvents)?;
31        Ok(BlockEvents(events.clone()))
32    }
33}
34
35impl<Ctx> FromJobCallParts<Ctx> for BlockEvents
36where
37    Ctx: Send + Sync,
38{
39    type Rejection = MissingBlockEvents;
40
41    async fn from_job_call_parts(
42        parts: &mut JobCallParts,
43        _: &Ctx,
44    ) -> Result<Self, Self::Rejection> {
45        Self::try_from(parts)
46    }
47}
48
49/// Extracts the first event of type `T` that happened in the current block.
50pub struct FirstEvent<T>(pub T);
51
52/// Extracts the last event of type `T` that happened in the current block.
53pub struct LastEvent<T>(pub T);
54
55/// Extracts all the events of type `T` that happened in the current block.
56pub struct Event<T>(pub Vec<T>);
57
58blueprint_core::__impl_deref!(FirstEvent);
59blueprint_core::__impl_deref!(LastEvent);
60
61define_rejection! {
62  #[body = "No event that matches this event found in the extensions. Did you forget to add the `AddBlockEventsLayer`?"]
63  /// A Rejection type for [`Event`] when it is missing in the Extensions.
64  /// This usually happens when the `AddBlockEventsLayer` is not added to the Router.
65  pub struct MissingEvent;
66}
67
68define_rejection! {
69  #[body = "An error occurred while trying to extract the event from subxt"]
70  /// A Rejection type for [`Event`] when an error occurs while trying to extract the event from subxt.
71  pub struct ClientError(Error);
72}
73
74composite_rejection! {
75    /// Rejection used for [`Event`].
76    ///
77    /// Contains one variant for each way the [`BlockHash`] extractor
78    /// can fail.
79    pub enum MissingEventRejection {
80        MissingBlockEvents,
81        MissingEvent,
82        ClientError,
83    }
84}
85
86macro_rules! impl_try_from {
87    ($t:ident, $method:ident) => {
88        impl<T> TryFrom<&mut JobCallParts> for $t<T>
89        where
90            T: StaticEvent,
91        {
92            type Error = MissingEventRejection;
93
94            fn try_from(parts: &mut JobCallParts) -> Result<Self, Self::Error> {
95                let event = parts
96                    .extensions
97                    .get::<Events<TangleConfig>>()
98                    .ok_or(MissingBlockEvents)?
99                    .$method::<T>()
100                    .map_err(ClientError::from_err)?
101                    .ok_or(MissingEvent)?;
102                Ok($t(event))
103            }
104        }
105    };
106}
107impl_try_from!(FirstEvent, find_first);
108impl_try_from!(LastEvent, find_last);
109
110impl<T> TryFrom<&mut JobCallParts> for Event<T>
111where
112    T: StaticEvent,
113{
114    type Error = MissingEventRejection;
115
116    fn try_from(parts: &mut JobCallParts) -> Result<Self, Self::Error> {
117        let events = parts
118            .extensions
119            .get::<Events<TangleConfig>>()
120            .ok_or(MissingBlockEvents)?
121            .find::<T>()
122            .collect::<Result<Vec<T>, _>>()
123            .map_err(ClientError::from_err)?;
124        Ok(Event(events))
125    }
126}
127
128macro_rules! impl_from_job_call_parts {
129    ($t:ident) => {
130        impl<Ctx, T> FromJobCallParts<Ctx> for $t<T>
131        where
132            Ctx: Send + Sync,
133            T: StaticEvent,
134        {
135            type Rejection = MissingEventRejection;
136
137            async fn from_job_call_parts(
138                parts: &mut JobCallParts,
139                _: &Ctx,
140            ) -> Result<Self, Self::Rejection> {
141                Self::try_from(parts)
142            }
143        }
144    };
145}
146
147impl_from_job_call_parts!(FirstEvent);
148impl_from_job_call_parts!(LastEvent);
149impl_from_job_call_parts!(Event);
blueprint_tangle_extra/extract/block_events.rs

blueprint_tangle_extra/extract/
block_events.rs
