sui_rpc/client/lists.rs
1use super::Client;
2use super::Result;
3use crate::proto::sui::rpc::v2::Balance;
4use crate::proto::sui::rpc::v2::DynamicField;
5use crate::proto::sui::rpc::v2::ListBalancesRequest;
6use crate::proto::sui::rpc::v2::ListDynamicFieldsRequest;
7use crate::proto::sui::rpc::v2::ListOwnedObjectsRequest;
8use crate::proto::sui::rpc::v2::ListPackageVersionsRequest;
9use crate::proto::sui::rpc::v2::Object;
10use crate::proto::sui::rpc::v2::PackageVersion;
11use futures::stream;
12use futures::stream::Stream;
13
14impl Client {
15 /// Creates a stream of objects based on the provided request.
16 ///
17 /// The stream handles pagination automatically by using the page_token from responses
18 /// to fetch subsequent pages. The original request's page_token is used as the starting point.
19 ///
20 /// # Arguments
21 /// * `request` - The initial `ListOwnedObjectsRequest` with search criteria
22 ///
23 /// # Returns
24 /// A stream that yields `Result<Object>` instances. If any RPC call fails, the
25 /// tonic::Status from that request is returned.
26 pub fn list_owned_objects(
27 &self,
28 request: impl tonic::IntoRequest<ListOwnedObjectsRequest>,
29 ) -> impl Stream<Item = Result<Object>> + 'static {
30 let client = self.clone();
31 let request = request.into_request();
32
33 stream::unfold(
34 (
35 Vec::new().into_iter(), // current batch of objects
36 true, // has_next_page
37 request, // request (page_token will be updated as we paginate)
38 client, // client for making requests
39 ),
40 move |(mut iter, mut has_next_page, mut request, mut client)| async move {
41 if let Some(item) = iter.next() {
42 return Some((Ok(item), (iter, has_next_page, request, client)));
43 }
44
45 // A page may be empty while still carrying a next_page_token
46 // (for example, a filtered scan that hit a server-side
47 // budget), so keep following the token until a page yields an
48 // item or pagination ends. Each response's token reflects
49 // server-side scan progress, so this terminates.
50 while has_next_page {
51 let new_request = tonic::Request::from_parts(
52 request.metadata().clone(),
53 request.extensions().clone(),
54 request.get_ref().clone(),
55 );
56
57 match client.state_client().list_owned_objects(new_request).await {
58 Ok(response) => {
59 let response = response.into_inner();
60 let mut iter = response.objects.into_iter();
61
62 has_next_page = response.next_page_token.is_some();
63 request.get_mut().page_token = response.next_page_token;
64
65 if let Some(item) = iter.next() {
66 return Some((Ok(item), (iter, has_next_page, request, client)));
67 }
68 }
69 Err(e) => {
70 // Return error and terminate stream
71 request.get_mut().page_token = None;
72 return Some((
73 Err(e),
74 (Vec::new().into_iter(), false, request, client),
75 ));
76 }
77 }
78 }
79 None
80 },
81 )
82 }
83
84 /// Creates a stream of `DynamicField`s based on the provided request.
85 ///
86 /// The stream handles pagination automatically by using the page_token from responses
87 /// to fetch subsequent pages. The original request's page_token is used as the starting point.
88 ///
89 /// # Arguments
90 /// * `request` - The initial `ListDynamicFieldsRequest` with search criteria
91 ///
92 /// # Returns
93 /// A stream that yields `Result<DynamicField>` instances. If any RPC call fails, the
94 /// tonic::Status from that request is returned.
95 pub fn list_dynamic_fields(
96 &self,
97 request: impl tonic::IntoRequest<ListDynamicFieldsRequest>,
98 ) -> impl Stream<Item = Result<DynamicField>> + 'static {
99 let client = self.clone();
100 let request = request.into_request();
101
102 stream::unfold(
103 (
104 Vec::new().into_iter(), // current batch of objects
105 true, // has_next_page
106 request, // request (page_token will be updated as we paginate)
107 client, // client for making requests
108 ),
109 move |(mut iter, mut has_next_page, mut request, mut client)| async move {
110 if let Some(item) = iter.next() {
111 return Some((Ok(item), (iter, has_next_page, request, client)));
112 }
113
114 // A page may be empty while still carrying a next_page_token
115 // (for example, a filtered scan that hit a server-side
116 // budget), so keep following the token until a page yields an
117 // item or pagination ends. Each response's token reflects
118 // server-side scan progress, so this terminates.
119 while has_next_page {
120 let new_request = tonic::Request::from_parts(
121 request.metadata().clone(),
122 request.extensions().clone(),
123 request.get_ref().clone(),
124 );
125
126 match client.state_client().list_dynamic_fields(new_request).await {
127 Ok(response) => {
128 let response = response.into_inner();
129 let mut iter = response.dynamic_fields.into_iter();
130
131 has_next_page = response.next_page_token.is_some();
132 request.get_mut().page_token = response.next_page_token;
133
134 if let Some(item) = iter.next() {
135 return Some((Ok(item), (iter, has_next_page, request, client)));
136 }
137 }
138 Err(e) => {
139 // Return error and terminate stream
140 request.get_mut().page_token = None;
141 return Some((
142 Err(e),
143 (Vec::new().into_iter(), false, request, client),
144 ));
145 }
146 }
147 }
148 None
149 },
150 )
151 }
152
153 /// Creates a stream of `Balance`s based on the provided request.
154 ///
155 /// The stream handles pagination automatically by using the page_token from responses
156 /// to fetch subsequent pages. The original request's page_token is used as the starting point.
157 ///
158 /// # Arguments
159 /// * `request` - The initial `ListBalancesRequest` with search criteria
160 ///
161 /// # Returns
162 /// A stream that yields `Result<Balance>` instances. If any RPC call fails, the
163 /// tonic::Status from that request is returned.
164 pub fn list_balances(
165 &self,
166 request: impl tonic::IntoRequest<ListBalancesRequest>,
167 ) -> impl Stream<Item = Result<Balance>> + 'static {
168 let client = self.clone();
169 let request = request.into_request();
170
171 stream::unfold(
172 (
173 Vec::new().into_iter(), // current batch of objects
174 true, // has_next_page
175 request, // request (page_token will be updated as we paginate)
176 client, // client for making requests
177 ),
178 move |(mut iter, mut has_next_page, mut request, mut client)| async move {
179 if let Some(item) = iter.next() {
180 return Some((Ok(item), (iter, has_next_page, request, client)));
181 }
182
183 // A page may be empty while still carrying a next_page_token
184 // (for example, a filtered scan that hit a server-side
185 // budget), so keep following the token until a page yields an
186 // item or pagination ends. Each response's token reflects
187 // server-side scan progress, so this terminates.
188 while has_next_page {
189 let new_request = tonic::Request::from_parts(
190 request.metadata().clone(),
191 request.extensions().clone(),
192 request.get_ref().clone(),
193 );
194
195 match client.state_client().list_balances(new_request).await {
196 Ok(response) => {
197 let response = response.into_inner();
198 let mut iter = response.balances.into_iter();
199
200 has_next_page = response.next_page_token.is_some();
201 request.get_mut().page_token = response.next_page_token;
202
203 if let Some(item) = iter.next() {
204 return Some((Ok(item), (iter, has_next_page, request, client)));
205 }
206 }
207 Err(e) => {
208 // Return error and terminate stream
209 request.get_mut().page_token = None;
210 return Some((
211 Err(e),
212 (Vec::new().into_iter(), false, request, client),
213 ));
214 }
215 }
216 }
217 None
218 },
219 )
220 }
221
222 /// Creates a stream of `PackageVersion`s based on the provided request.
223 ///
224 /// The stream handles pagination automatically by using the page_token from responses
225 /// to fetch subsequent pages. The original request's page_token is used as the starting point.
226 ///
227 /// # Arguments
228 /// * `request` - The initial `ListPackageVersionsRequest` with search criteria
229 ///
230 /// # Returns
231 /// A stream that yields `Result<PackageVersion>` instances. If any RPC call fails, the
232 /// tonic::Status from that request is returned.
233 pub fn list_package_versions(
234 &self,
235 request: impl tonic::IntoRequest<ListPackageVersionsRequest>,
236 ) -> impl Stream<Item = Result<PackageVersion>> + 'static {
237 let client = self.clone();
238 let request = request.into_request();
239
240 stream::unfold(
241 (
242 Vec::new().into_iter(), // current batch of objects
243 true, // has_next_page
244 request, // request (page_token will be updated as we paginate)
245 client, // client for making requests
246 ),
247 move |(mut iter, mut has_next_page, mut request, mut client)| async move {
248 if let Some(item) = iter.next() {
249 return Some((Ok(item), (iter, has_next_page, request, client)));
250 }
251
252 // A page may be empty while still carrying a next_page_token
253 // (for example, a filtered scan that hit a server-side
254 // budget), so keep following the token until a page yields an
255 // item or pagination ends. Each response's token reflects
256 // server-side scan progress, so this terminates.
257 while has_next_page {
258 let new_request = tonic::Request::from_parts(
259 request.metadata().clone(),
260 request.extensions().clone(),
261 request.get_ref().clone(),
262 );
263
264 match client
265 .package_client()
266 .list_package_versions(new_request)
267 .await
268 {
269 Ok(response) => {
270 let response = response.into_inner();
271 let mut iter = response.versions.into_iter();
272
273 has_next_page = response.next_page_token.is_some();
274 request.get_mut().page_token = response.next_page_token;
275
276 if let Some(item) = iter.next() {
277 return Some((Ok(item), (iter, has_next_page, request, client)));
278 }
279 }
280 Err(e) => {
281 // Return error and terminate stream
282 request.get_mut().page_token = None;
283 return Some((
284 Err(e),
285 (Vec::new().into_iter(), false, request, client),
286 ));
287 }
288 }
289 }
290 None
291 },
292 )
293 }
294}