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
25/// The receipt callback accepts only positions committed to local storage.
26pub struct IncomingSubscription<E> {
27    /// Ordered transport events. Reading an event does not commit receipt.
28    pub events: BoxDynStream<'static, Result<IncomingEvent, E>>,
29    receipt: Box<dyn ReceiptSink>,
30}
31
32impl<E> IncomingSubscription<E> {
33    /// Attach a receipt callback to an owned event stream.
34    pub fn new(
35        events: BoxDynStream<'static, Result<IncomingEvent, E>>,
36        receipt: impl Fn(TopicCursor) + MaybeSend + MaybeSync + 'static,
37    ) -> Self {
38        Self {
39            events,
40            receipt: Box::new(receipt),
41        }
42    }
43
44    /// Advance resume positions only to the committed received prefix `F`.
45    /// Do not acknowledge positions that exist only in memory.
46    pub fn acknowledge_received(&self, cursors: TopicCursor) {
47        self.receipt.acknowledge(cursors);
48    }
49
50    /// Convert transport errors without changing the receipt callback.
51    pub fn map_error<U, F>(self, map: F) -> IncomingSubscription<U>
52    where
53        E: 'static,
54        U: 'static,
55        F: Fn(E) -> U + MaybeSend + 'static,
56    {
57        IncomingSubscription {
58            events: Box::pin(self.events.map(move |event| event.map_err(&map))),
59            receipt: self.receipt,
60        }
61    }
62}
63
64/// One ordered delivery from a topic. `after` is the cursor used for this read.
65#[derive(Clone, Debug)]
66pub struct OrderedEnvelopeBatch {
67    /// The single topic shared by every envelope in this batch.
68    pub topic: Topic,
69    /// Read position before this batch, not a durable receipt acknowledgement.
70    pub after: Cursor,
71    /// Envelopes in strictly increasing sequence order. Sequence gaps are valid.
72    pub envelopes: Vec<ServerEnvelope>,
73}
74
75/// Transport events do not mean that an envelope is stored or processed.
76#[derive(Clone, Debug)]
77pub enum IncomingEvent {
78    /// Fixed catch-up targets for the topics in one accepted registration.
79    Registered {
80        /// Read positions used when these topics were registered.
81        starts: TopicCursor,
82        /// Targets captured at registration. Later messages do not raise them.
83        targets: TopicCursor,
84    },
85    /// A validated transport batch whose raw bytes still need durable receipt.
86    OrderedBatch(OrderedEnvelopeBatch),
87    /// The feed ended. Reopen from committed receipt positions.
88    Disconnected,
89}
xmtp_proto/types/incoming_event.rs

xmtp_proto/types/
incoming_event.rs
