xmtp_proto/types/
incoming_event.rs

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

xmtp_proto/types/
incoming_event.rs
