sui_rpc/client/ledger_streams/
facade.rs

1use futures::Stream;
2
3use super::super::Client;
4use super::super::Result;
5use super::adapter::CheckpointAdapter;
6use super::adapter::EventAdapter;
7use super::adapter::SubscriptionAdapter;
8use super::adapter::TransactionAdapter;
9use super::list::ListDriver;
10use super::stream::Driver;
11use super::stream::Start;
12use super::types::CheckpointStreamFrame;
13use super::types::CheckpointStreamRequest;
14use super::types::CheckpointStreamStart;
15use super::types::EventStreamFrame;
16use super::types::EventStreamRequest;
17use super::types::EventStreamStart;
18use super::types::LedgerStreamConfig;
19use super::types::ListConfig;
20use super::types::TransactionStreamFrame;
21use super::types::TransactionStreamRequest;
22use super::types::TransactionStreamStart;
23use crate::proto::sui::rpc::v2::ListCheckpointsRequest;
24use crate::proto::sui::rpc::v2::ListCheckpointsResponse;
25use crate::proto::sui::rpc::v2::ListEventsRequest;
26use crate::proto::sui::rpc::v2::ListEventsResponse;
27use crate::proto::sui::rpc::v2::ListTransactionsRequest;
28use crate::proto::sui::rpc::v2::ListTransactionsResponse;
29use crate::proto::sui::rpc::v2::SubscribeCheckpointsRequest;
30use crate::proto::sui::rpc::v2::SubscribeEventsRequest;
31use crate::proto::sui::rpc::v2::SubscribeTransactionsRequest;
32
33impl Client {
34    /// Paginates `ListCheckpoints` requests into a stream of raw response pages.
35    ///
36    /// Automatically follows `ItemLimit` and `ScanLimit` pagination until reaching the
37    /// request's end bound or the ledger tip.
38    pub fn list_checkpoints(
39        &self,
40        request: ListCheckpointsRequest,
41    ) -> impl Stream<Item = Result<ListCheckpointsResponse>> + Send + 'static {
42        self.list_checkpoints_with_config(request, ListConfig::default())
43    }
44    /// Performs [`Client::list_checkpoints`] using `config`.
45    pub fn list_checkpoints_with_config(
46        &self,
47        request: ListCheckpointsRequest,
48        config: ListConfig,
49    ) -> impl Stream<Item = Result<ListCheckpointsResponse>> + Send + 'static {
50        list_stream(ListDriver::<CheckpointAdapter>::new(
51            self.clone(),
52            request,
53            config.into_stream_config(),
54        ))
55    }
56
57    /// Streams checkpoints indefinitely, automatically retrying transient errors.
58    ///
59    /// The request's read mask must include `sequence_number` (or `*`). To resume after a
60    /// previous checkpoint, pass `cursor + 1` to [`CheckpointStreamStart::Checkpoint`].
61    ///
62    /// For a finite read that stops at a bound, use [`Client::list_checkpoints`].
63    pub fn stream_checkpoints(
64        &self,
65        request: CheckpointStreamRequest,
66    ) -> impl Stream<Item = Result<CheckpointStreamFrame>> + Send + 'static {
67        self.stream_checkpoints_with_config(request, LedgerStreamConfig::default())
68    }
69
70    /// Performs [`Client::stream_checkpoints`] using `config`.
71    pub fn stream_checkpoints_with_config(
72        &self,
73        request: CheckpointStreamRequest,
74        config: LedgerStreamConfig,
75    ) -> impl Stream<Item = Result<CheckpointStreamFrame>> + Send + 'static {
76        let subscribe_payload = SubscribeCheckpointsRequest {
77            read_mask: request.read_mask.clone(),
78            filter: request.filter.clone(),
79        };
80        let list_template = ListCheckpointsRequest {
81            read_mask: request.read_mask,
82            filter: request.filter,
83            start_checkpoint: None,
84            end_checkpoint: None,
85            options: None,
86        };
87        let start = match request.start {
88            CheckpointStreamStart::Tip => Start::Tip,
89            CheckpointStreamStart::Checkpoint(checkpoint) => Start::Checkpoint(checkpoint),
90        };
91        resumable_stream(Driver::<CheckpointAdapter>::new_stream(
92            self.clone(),
93            subscribe_payload,
94            list_template,
95            start,
96            request.delivery,
97            config,
98        ))
99    }
100
101    /// Paginates `ListTransactions` requests into a stream of raw response pages.
102    ///
103    /// Automatically follows `ItemLimit` and `ScanLimit` pagination until reaching the
104    /// request's end bound or the ledger tip.
105    pub fn list_transactions(
106        &self,
107        request: ListTransactionsRequest,
108    ) -> impl Stream<Item = Result<ListTransactionsResponse>> + Send + 'static {
109        self.list_transactions_with_config(request, ListConfig::default())
110    }
111    /// Performs [`Client::list_transactions`] using `config`.
112    pub fn list_transactions_with_config(
113        &self,
114        request: ListTransactionsRequest,
115        config: ListConfig,
116    ) -> impl Stream<Item = Result<ListTransactionsResponse>> + Send + 'static {
117        list_stream(ListDriver::<TransactionAdapter>::new(
118            self.clone(),
119            request,
120            config.into_stream_config(),
121        ))
122    }
123
124    /// Streams transactions indefinitely, automatically retrying transient errors.
125    ///
126    /// The request's read mask must include `checkpoint` and `transaction_index` (or `*`).
127    /// To resume from a previous frame, pass `frame.cursor` to [`TransactionStreamStart::Resume`].
128    ///
129    /// For a finite read that stops at a bound, use [`Client::list_transactions`].
130    pub fn stream_transactions(
131        &self,
132        request: TransactionStreamRequest,
133    ) -> impl Stream<Item = Result<TransactionStreamFrame>> + Send + 'static {
134        self.stream_transactions_with_config(request, LedgerStreamConfig::default())
135    }
136
137    /// Performs [`Client::stream_transactions`] using `config`.
138    pub fn stream_transactions_with_config(
139        &self,
140        request: TransactionStreamRequest,
141        config: LedgerStreamConfig,
142    ) -> impl Stream<Item = Result<TransactionStreamFrame>> + Send + 'static {
143        let subscribe_payload = SubscribeTransactionsRequest {
144            read_mask: request.read_mask.clone(),
145            filter: request.filter.clone(),
146        };
147        let list_template = ListTransactionsRequest {
148            read_mask: request.read_mask,
149            filter: request.filter,
150            start_checkpoint: None,
151            end_checkpoint: None,
152            options: None,
153        };
154        let start = match request.start {
155            TransactionStreamStart::Tip => Start::Tip,
156            TransactionStreamStart::Checkpoint(checkpoint) => Start::Checkpoint(checkpoint),
157            TransactionStreamStart::Resume(cursor) => Start::After(cursor),
158        };
159        resumable_stream(Driver::<TransactionAdapter>::new_stream(
160            self.clone(),
161            subscribe_payload,
162            list_template,
163            start,
164            request.delivery,
165            config,
166        ))
167    }
168
169    /// Paginates `ListEvents` requests into a stream of raw response pages.
170    ///
171    /// Automatically follows `ItemLimit` and `ScanLimit` pagination until reaching the
172    /// request's end bound or the ledger tip.
173    pub fn list_events(
174        &self,
175        request: ListEventsRequest,
176    ) -> impl Stream<Item = Result<ListEventsResponse>> + Send + 'static {
177        self.list_events_with_config(request, ListConfig::default())
178    }
179    /// Performs [`Client::list_events`] using `config`.
180    pub fn list_events_with_config(
181        &self,
182        request: ListEventsRequest,
183        config: ListConfig,
184    ) -> impl Stream<Item = Result<ListEventsResponse>> + Send + 'static {
185        list_stream(ListDriver::<EventAdapter>::new(
186            self.clone(),
187            request,
188            config.into_stream_config(),
189        ))
190    }
191
192    /// Streams events indefinitely, automatically retrying transient errors.
193    ///
194    /// The request's read mask must include `checkpoint`, `transaction_index`, and `event_index`
195    /// (or `*`). To resume from a previous frame, pass `frame.cursor` to
196    /// [`EventStreamStart::Resume`].
197    ///
198    /// For a finite read that stops at a bound, use [`Client::list_events`].
199    pub fn stream_events(
200        &self,
201        request: EventStreamRequest,
202    ) -> impl Stream<Item = Result<EventStreamFrame>> + Send + 'static {
203        self.stream_events_with_config(request, LedgerStreamConfig::default())
204    }
205
206    /// Performs [`Client::stream_events`] using `config`.
207    pub fn stream_events_with_config(
208        &self,
209        request: EventStreamRequest,
210        config: LedgerStreamConfig,
211    ) -> impl Stream<Item = Result<EventStreamFrame>> + Send + 'static {
212        let subscribe_payload = SubscribeEventsRequest {
213            read_mask: request.read_mask.clone(),
214            filter: request.filter.clone(),
215        };
216        let list_template = ListEventsRequest {
217            read_mask: request.read_mask,
218            filter: request.filter,
219            start_checkpoint: None,
220            end_checkpoint: None,
221            options: None,
222        };
223        let start = match request.start {
224            EventStreamStart::Tip => Start::Tip,
225            EventStreamStart::Checkpoint(checkpoint) => Start::Checkpoint(checkpoint),
226            EventStreamStart::Resume(cursor) => Start::After(cursor),
227        };
228        resumable_stream(Driver::<EventAdapter>::new_stream(
229            self.clone(),
230            subscribe_payload,
231            list_template,
232            start,
233            request.delivery,
234            config,
235        ))
236    }
237}
238
239fn list_stream<A: SubscriptionAdapter>(
240    driver: ListDriver<A>,
241) -> impl Stream<Item = Result<A::ListResponse>> + Send + 'static {
242    futures::stream::unfold(driver, |mut driver| async move {
243        driver.next().await.map(|item| (item, driver))
244    })
245}
246
247fn resumable_stream<A: SubscriptionAdapter>(
248    driver: Driver<A>,
249) -> impl Stream<Item = Result<A::Output>> + Send + 'static {
250    futures::stream::unfold(driver, |mut driver| async move {
251        driver.next().await.map(|item| (item, driver))
252    })
253}
sui_rpc/client/ledger_streams/facade.rs

sui_rpc/client/ledger_streams/
facade.rs
