xmtp_proto/types/
incoming_event.rs

1use super::{Cursor, ReceivedEnvelope, Topic, TopicCursor};
2use futures::StreamExt;
3use xmtp_common::{BoxDynStream, MaybeSend, MaybeSync};
4
5/// Limits for one complete transport delivery before payload processing.
6#[derive(Clone, Copy, Debug)]
7pub struct IncomingBatchLimits {
8    /// Maximum number of envelopes across all topics in the delivery.
9    pub max_rows: usize,
10    /// Maximum sum of received `ServerEnvelope` sizes in the delivery.
11    pub max_bytes: usize,
12}
13
14impl IncomingBatchLimits {
15    /// One fetched page of the default client stream policy, for a reader
16    /// that holds no client policy.
17    pub const STREAM_BATCH: Self = Self {
18        max_rows: xmtp_configuration::STREAM_BATCH_ROWS as usize,
19        max_bytes: xmtp_configuration::STREAM_BATCH_BYTES as usize,
20    };
21}
22
23trait ReceiptSink: MaybeSend + MaybeSync {
24    fn acknowledge(&self, cursors: TopicCursor);
25}
26
27impl<F: Fn(TopicCursor) + MaybeSend + MaybeSync> ReceiptSink for F {
28    fn acknowledge(&self, cursors: TopicCursor) {
29        self(cursors);
30    }
31}
32
33trait ConnectionSource: MaybeSend + MaybeSync {
34    fn is_connected(&self) -> bool;
35}
36
37impl<F: Fn() -> bool + MaybeSend + MaybeSync> ConnectionSource for F {
38    fn is_connected(&self) -> bool {
39        self()
40    }
41}
42
43/// The receipt callback accepts only positions committed to local storage.
44pub struct IncomingSubscription<E> {
45    /// Ordered transport events. Reading an event does not commit receipt.
46    pub events: BoxDynStream<'static, Result<IncomingEvent, E>>,
47    receipt: Box<dyn ReceiptSink>,
48    connection: Option<Box<dyn ConnectionSource>>,
49}
50
51impl<E> IncomingSubscription<E> {
52    /// Attach a receipt callback to an owned event stream.
53    pub fn new(
54        events: BoxDynStream<'static, Result<IncomingEvent, E>>,
55        receipt: impl Fn(TopicCursor) + MaybeSend + MaybeSync + 'static,
56    ) -> Self {
57        Self {
58            events,
59            receipt: Box::new(receipt),
60            connection: None,
61        }
62    }
63
64    /// Bind a retained subscription to its current network receipt source.
65    pub fn with_connection_source(
66        mut self,
67        connection: impl Fn() -> bool + MaybeSend + MaybeSync + 'static,
68    ) -> Self {
69        self.connection = Some(Box::new(connection));
70        self
71    }
72
73    /// A retained lease can remain open while its network wire is unavailable.
74    pub fn is_connected(&self) -> bool {
75        self.connection
76            .as_ref()
77            .is_none_or(|connection| connection.is_connected())
78    }
79
80    /// Advance resume positions only to the committed received prefix `F`.
81    /// Do not acknowledge positions that exist only in memory.
82    pub fn acknowledge_received(&self, cursors: TopicCursor) {
83        self.receipt.acknowledge(cursors);
84    }
85
86    /// Convert transport errors without changing the receipt callback.
87    pub fn map_error<U, F>(self, map: F) -> IncomingSubscription<U>
88    where
89        E: 'static,
90        U: 'static,
91        F: Fn(E) -> U + MaybeSend + 'static,
92    {
93        IncomingSubscription {
94            events: Box::pin(self.events.map(move |event| event.map_err(&map))),
95            receipt: self.receipt,
96            connection: self.connection,
97        }
98    }
99}
100
101/// One ordered delivery from a topic. `after` is the cursor used for this read.
102#[derive(Clone, Debug)]
103pub struct OrderedEnvelopeBatch {
104    /// The single topic shared by every envelope in this batch.
105    pub topic: Topic,
106    /// Read position before this batch, not a durable receipt acknowledgement.
107    pub after: Cursor,
108    /// Envelopes in strictly increasing sequence order. Sequence gaps are valid.
109    pub envelopes: Vec<ReceivedEnvelope>,
110}
111
112/// Transport events do not mean that an envelope is stored or processed.
113#[derive(Clone, Debug)]
114pub enum IncomingEvent {
115    /// Fixed catch-up targets for the topics in one accepted registration.
116    Registered {
117        /// Read positions used when these topics were registered.
118        starts: TopicCursor,
119        /// Targets captured at registration. Later messages do not raise them.
120        targets: TopicCursor,
121    },
122    /// A validated transport batch whose raw bytes still need durable receipt.
123    OrderedBatch(OrderedEnvelopeBatch),
124    /// The feed ended. Reopen from committed receipt positions.
125    Disconnected,
126}
xmtp_proto/types/incoming_event.rs

xmtp_proto/types/
incoming_event.rs
