ibapi/accounts/sync.rs
1//! Synchronous implementation of account management functionality
2
3use std::sync::Arc;
4use std::time::Duration;
5
6use time::OffsetDateTime;
7
8use crate::client::blocking::{ClientRequestBuilders, Subscription, SubscriptionBuilderExt};
9use crate::common::request_helpers::{self, empty_on_end_of_stream, expect_proto};
10use crate::messages::OutgoingMessages;
11use crate::{client::sync::Client, Error};
12
13use super::common::snapshots::SnapshotBuilder;
14use super::common::{decoders, encoders};
15use super::types::{AccountGroup, AccountId, ContractId, ModelCode};
16use super::*;
17
18impl Client {
19 /// TWS's current time. TWS is synchronized with the server (not local computer) using NTP and this function will receive the current time in TWS.
20 ///
21 /// # Errors
22 /// [`Error::Timeout`] if TWS sends no answer within 30 seconds, or within [`one_shot_timeout`](crate::client::blocking::ClientBuilder::one_shot_timeout) when set.
23 ///
24 /// # Examples
25 ///
26 /// ```no_run
27 /// use ibapi::client::blocking::Client;
28 ///
29 /// let client = Client::connect("127.0.0.1:4002", 100).expect("connection failed");
30 /// let server_time = client.server_time().expect("error requesting server time");
31 /// println!("server time: {server_time:?}");
32 /// ```
33 pub fn server_time(&self) -> Result<OffsetDateTime, Error> {
34 request_helpers::blocking::one_shot_shared(
35 self,
36 OutgoingMessages::RequestCurrentTime,
37 encoders::encode_request_server_time,
38 expect_proto(decoders::decode_server_time_proto),
39 )
40 }
41
42 /// Requests the current server time with millisecond precision.
43 ///
44 /// # Errors
45 /// [`Error::Timeout`] if TWS sends no answer within 30 seconds, or within [`one_shot_timeout`](crate::client::blocking::ClientBuilder::one_shot_timeout) when set.
46 ///
47 /// # Examples
48 ///
49 /// ```no_run
50 /// use ibapi::client::blocking::Client;
51 ///
52 /// let client = Client::connect("127.0.0.1:4002", 100).expect("connection failed");
53 /// let server_time = client.server_time_millis().expect("error requesting server time");
54 /// println!("server time to the millisecond: {server_time:?}");
55 /// ```
56 pub fn server_time_millis(&self) -> Result<OffsetDateTime, Error> {
57 request_helpers::blocking::one_shot_shared(
58 self,
59 OutgoingMessages::RequestCurrentTimeInMillis,
60 encoders::encode_request_server_time_millis,
61 expect_proto(decoders::decode_server_time_millis_proto),
62 )
63 }
64
65 /// Subscribes to [PositionUpdate]s for all accessible accounts.
66 /// All positions sent initially, and then only updates as positions change.
67 ///
68 /// # Examples
69 ///
70 /// ```no_run
71 /// use ibapi::client::blocking::Client;
72 /// use ibapi::accounts::PositionUpdate;
73 ///
74 /// let client = Client::connect("127.0.0.1:4002", 100).expect("connection failed");
75 /// let subscription = client.positions().expect("error requesting positions");
76 /// for position_response in subscription.iter_data() {
77 /// match position_response? {
78 /// PositionUpdate::Position(position) => println!("{position:?}"),
79 /// PositionUpdate::PositionEnd => println!("initial set of positions received"),
80 /// }
81 /// }
82 /// # Ok::<(), ibapi::Error>(())
83 /// ```
84 pub fn positions(&self) -> Result<Subscription<PositionUpdate>, Error> {
85 let request = encoders::encode_request_positions()?;
86 self.subscription().send_shared(OutgoingMessages::RequestPositions, request)
87 }
88
89 /// Subscribes to [PositionUpdateMulti] updates for account and/or model.
90 /// Initially all positions are returned, and then updates are returned for any position changes in real time.
91 ///
92 /// # Arguments
93 /// * `account` - If an account Id is provided, only the account's positions belonging to the specified model will be delivered.
94 /// * `model_code` - The code of the model's positions we are interested in.
95 ///
96 /// # Examples
97 ///
98 /// ```no_run
99 /// use ibapi::client::blocking::Client;
100 ///
101 /// let client = Client::connect("127.0.0.1:4002", 100).expect("connection failed");
102 ///
103 /// use ibapi::accounts::types::AccountId;
104 ///
105 /// let account = AccountId("U1234567".to_string());
106 /// let subscription = client.positions_multi(Some(&account), None).expect("error requesting positions by model");
107 /// for position in subscription.iter_data() {
108 /// match position {
109 /// Ok(position) => println!("{position:?}"),
110 /// Err(e) => eprintln!("error: {e}"),
111 /// }
112 /// }
113 /// ```
114 pub fn positions_multi(&self, account: Option<&AccountId>, model_code: Option<&ModelCode>) -> Result<Subscription<PositionUpdateMulti>, Error> {
115 let builder = self.request();
116 let request = encoders::encode_request_positions_multi(builder.request_id(), account, model_code)?;
117
118 builder.send(request)
119 }
120
121 /// Creates subscription for real time daily PnL and unrealized PnL updates.
122 ///
123 /// # Arguments
124 /// * `account` - account for which to receive PnL updates
125 /// * `model_code` - specify to request PnL updates for a specific model
126 ///
127 /// # Examples
128 ///
129 /// ```no_run
130 /// use ibapi::client::blocking::Client;
131 ///
132 /// let client = Client::connect("127.0.0.1:4002", 100).expect("connection failed");
133 /// use ibapi::accounts::types::AccountId;
134 ///
135 /// let account = AccountId("account id".to_string());
136 /// let subscription = client.pnl(&account, None).expect("error requesting pnl");
137 /// for pnl in subscription.iter_data() {
138 /// match pnl {
139 /// Ok(pnl) => println!("{pnl:?}"),
140 /// Err(e) => eprintln!("error: {e}"),
141 /// }
142 /// }
143 /// ```
144 pub fn pnl(&self, account: &AccountId, model_code: Option<&ModelCode>) -> Result<Subscription<PnL>, Error> {
145 let builder = self.request();
146 let request = encoders::encode_request_pnl(builder.request_id(), account, model_code)?;
147 builder.send(request)
148 }
149
150 /// Requests real time updates for daily PnL of individual positions.
151 ///
152 /// # Arguments
153 /// * `account` - Account in which position exists
154 /// * `contract_id` - Contract ID of contract to receive daily PnL updates for. Note: does not return response if invalid conId is entered.
155 /// * `model_code` - Model in which position exists
156 ///
157 /// # Examples
158 ///
159 /// ```no_run
160 /// use ibapi::client::blocking::Client;
161 ///
162 /// let client = Client::connect("127.0.0.1:4002", 100).expect("connection failed");
163 ///
164 /// use ibapi::accounts::types::{AccountId, ContractId};
165 ///
166 /// let account = AccountId("<account id>".to_string());
167 /// let contract_id = ContractId(1001);
168 ///
169 /// let subscription = client.pnl_single(&account, contract_id, None).expect("error requesting pnl");
170 /// for pnl in subscription.iter_data() {
171 /// match pnl {
172 /// Ok(pnl) => println!("{pnl:?}"),
173 /// Err(e) => eprintln!("error: {e}"),
174 /// }
175 /// }
176 /// ```
177 pub fn pnl_single(&self, account: &AccountId, contract_id: ContractId, model_code: Option<&ModelCode>) -> Result<Subscription<PnLSingle>, Error> {
178 let builder = self.request();
179 let request = encoders::encode_request_pnl_single(builder.request_id(), account, contract_id, model_code)?;
180 builder.send(request)
181 }
182
183 /// Requests a specific account's summary. Subscribes to the account summary as presented in the TWS' Account Summary tab. Data received is specified by using a specific tags value.
184 ///
185 /// # Subscription lifetime
186 ///
187 /// The subscription stays open until it is dropped or cancelled. TWS first sends the requested
188 /// tags followed by [`AccountSummaryResult::End`], then keeps pushing
189 /// [`AccountSummaryResult::Summary`] rows without another `End`: when values change, and
190 /// periodically, when a push can resend rows whose values did not change. Dropping the subscription cancels
191 /// the request.
192 ///
193 /// # Arguments
194 /// * `group` - Set to "All" to return account summary data for all accounts, or set to a specific Advisor Account Group name that has already been created in TWS Global Configuration.
195 /// * `tags` - List of the desired tags.
196 ///
197 /// # Examples
198 ///
199 /// ```no_run
200 /// use ibapi::client::blocking::Client;
201 /// use ibapi::accounts::AccountSummaryTags;
202 /// use ibapi::accounts::types::AccountGroup;
203 ///
204 /// let client = Client::connect("127.0.0.1:4002", 100).expect("connection failed");
205 ///
206 /// let group = AccountGroup("All".to_string());
207 ///
208 /// let subscription = client.account_summary(&group, &[AccountSummaryTags::ACCOUNT_TYPE]).expect("error requesting account summary");
209 /// for summary in subscription.iter_data() {
210 /// match summary {
211 /// Ok(summary) => println!("{summary:?}"),
212 /// Err(e) => eprintln!("error: {e}"),
213 /// }
214 /// }
215 /// ```
216 pub fn account_summary(&self, group: &AccountGroup, tags: &[&str]) -> Result<Subscription<AccountSummaryResult>, Error> {
217 let builder = self.request();
218 let request = encoders::encode_request_account_summary(builder.request_id(), group, tags)?;
219 builder.send(request)
220 }
221
222 /// Subscribe to account summary updates as complete snapshots.
223 ///
224 /// Wraps [`account_summary`](Self::account_summary) and keeps the latest value of every row, so
225 /// each [`AccountSummarySnapshot`] holds the whole account rather than only the rows TWS pushed
226 /// last. TWS sends one `End` marker after the initial snapshot and none after the rows it pushes
227 /// later, so the first snapshot is emitted at the first `End`, even when empty, and each later
228 /// one once no row has arrived for `quiet`. Pushes arrive when values change and periodically
229 /// (IB documents every three minutes), and a push can resend unchanged rows. A later snapshot is
230 /// emitted only when a row changed a value since the previous one. Choose `quiet` longer than
231 /// the gap between the rows of one push, which arrive within milliseconds of each other.
232 ///
233 /// To fold the rows with your own batching instead, use [`AccountSummarySnapshot::apply`].
234 ///
235 /// Notices that arrive on the subscription are logged at `warn!`. For other streams that dump
236 /// then push, batch with [`Subscription::next_batch`](crate::subscriptions::Subscription::next_batch).
237 ///
238 /// Dropping the returned iterator cancels the subscription.
239 ///
240 /// # Arguments
241 /// * `group` - Set to "All" to return account summary data for all accounts, or set to a specific Advisor Account Group name that has already been created in TWS Global Configuration.
242 /// * `tags` - List of the desired tags.
243 /// * `quiet` - How long without a row completes a pushed update.
244 ///
245 /// # Examples
246 ///
247 /// ```no_run
248 /// use ibapi::client::blocking::Client;
249 /// use ibapi::accounts::AccountSummaryTags;
250 /// use ibapi::accounts::types::AccountGroup;
251 /// use std::time::Duration;
252 ///
253 /// let client = Client::connect("127.0.0.1:4002", 100).expect("connection failed");
254 ///
255 /// let group = AccountGroup("All".to_string());
256 /// let tags = &[AccountSummaryTags::NET_LIQUIDATION];
257 ///
258 /// let snapshots = client
259 /// .account_summary_snapshots(&group, tags, Duration::from_secs(1))
260 /// .expect("error requesting account summary");
261 /// for snapshot in snapshots {
262 /// let snapshot = snapshot.expect("account summary error");
263 /// println!("{} rows", snapshot.len());
264 /// }
265 /// ```
266 pub fn account_summary_snapshots(&self, group: &AccountGroup, tags: &[&str], quiet: Duration) -> Result<AccountSummarySnapshots, Error> {
267 let subscription = self.account_summary(group, tags)?;
268
269 Ok(AccountSummarySnapshots {
270 subscription,
271 builder: SnapshotBuilder::default(),
272 quiet,
273 })
274 }
275
276 /// Subscribes to a specific account's information and portfolio.
277 ///
278 /// All account values and positions will be returned initially, and then there will only be updates when there is a change in a position, or to an account value every 3 minutes if it has changed.
279 ///
280 /// TWS streams one account at a time per connection. Subscribing to the same
281 /// account again shares the stream; another account is refused until every
282 /// subscription to the first has been cancelled or dropped. To stream several
283 /// accounts at once, use [`account_updates_multi`](Client::account_updates_multi).
284 ///
285 /// # Arguments
286 /// * `account` - The account id (i.e. U1234567) for which the information is requested.
287 ///
288 /// # Errors
289 /// [`Error::AccountUpdatesInUse`] while another account's updates are live.
290 ///
291 /// # Examples
292 ///
293 /// ```no_run
294 /// use ibapi::client::blocking::Client;
295 /// use ibapi::accounts::AccountUpdate;
296 ///
297 /// let client = Client::connect("127.0.0.1:4002", 100).expect("connection failed");
298 ///
299 /// use ibapi::accounts::types::AccountId;
300 ///
301 /// let account = AccountId("U1234567".to_string());
302 ///
303 /// let subscription = client.account_updates(&account).expect("error requesting account updates");
304 /// for update in subscription.iter_data() {
305 /// let update = update?;
306 /// println!("{update:?}");
307 ///
308 /// // stop after full initial update
309 /// if let AccountUpdate::End = update {
310 /// subscription.cancel();
311 /// }
312 /// }
313 /// # Ok::<(), ibapi::Error>(())
314 /// ```
315 pub fn account_updates(&self, account: &AccountId) -> Result<Subscription<AccountUpdate>, Error> {
316 let request = encoders::encode_request_account_updates(true, account)?;
317 let subscription = self.message_bus.send_account_updates_request(account, &request)?;
318 Ok(Subscription::new(Arc::clone(&self.message_bus), subscription, self.decoder_context()))
319 }
320
321 /// Requests account updates for account and/or model.
322 ///
323 /// All account values and positions will be returned initially, and then there will only be updates when there is a change in a position, or to an account value every 3 minutes if it has changed.
324 /// Each subscription is routed by its own request id, so several accounts or models can stream at once.
325 ///
326 /// # Arguments
327 /// * `account` - Account values can be requested for a particular account.
328 /// * `model_code` - Account values can also be requested for a model.
329 ///
330 /// # Examples
331 ///
332 /// ```no_run
333 /// use ibapi::client::blocking::Client;
334 /// use ibapi::accounts::AccountUpdateMulti;
335 ///
336 /// let client = Client::connect("127.0.0.1:4002", 100).expect("connection failed");
337 ///
338 /// use ibapi::accounts::types::AccountId;
339 ///
340 /// let account = AccountId("U1234567".to_string());
341 ///
342 /// let subscription = client.account_updates_multi(Some(&account), None).expect("error requesting account updates multi");
343 /// for update in subscription.iter_data() {
344 /// let update = update?;
345 /// println!("{update:?}");
346 ///
347 /// // stop after full initial update
348 /// if let AccountUpdateMulti::End = update {
349 /// subscription.cancel();
350 /// }
351 /// }
352 /// # Ok::<(), ibapi::Error>(())
353 /// ```
354 pub fn account_updates_multi(
355 &self,
356 account: Option<&AccountId>,
357 model_code: Option<&ModelCode>,
358 ) -> Result<Subscription<AccountUpdateMulti>, Error> {
359 let builder = self.request();
360 let request = encoders::encode_request_account_updates_multi(builder.request_id(), account, model_code)?;
361
362 builder.send(request)
363 }
364
365 /// Fetch the list of accounts accessible to the current user.
366 ///
367 /// For the list TWS sent at connect, with no round trip, see
368 /// [`handshake_managed_accounts`](Self::handshake_managed_accounts).
369 ///
370 /// # Errors
371 /// [`Error::Timeout`] if TWS sends no answer within 30 seconds, or within [`one_shot_timeout`](crate::client::blocking::ClientBuilder::one_shot_timeout) when set.
372 ///
373 /// # Examples
374 ///
375 /// ```no_run
376 /// use ibapi::client::blocking::Client;
377 ///
378 /// let client = Client::connect("127.0.0.1:4002", 100).expect("connection failed");
379 ///
380 /// let accounts = client.managed_accounts().expect("error requesting managed accounts");
381 /// println!("managed accounts: {accounts:?}")
382 /// ```
383 pub fn managed_accounts(&self) -> Result<Vec<String>, Error> {
384 request_helpers::blocking::one_shot_shared(
385 self,
386 OutgoingMessages::RequestManagedAccounts,
387 encoders::encode_request_managed_accounts,
388 expect_proto(decoders::decode_managed_accounts_proto),
389 )
390 .or_else(empty_on_end_of_stream)
391 }
392
393 /// Get current [FamilyCode]s for all accessible accounts.
394 ///
395 /// # Errors
396 /// [`Error::Timeout`] if TWS sends no answer within 30 seconds, or within [`one_shot_timeout`](crate::client::blocking::ClientBuilder::one_shot_timeout) when set.
397 ///
398 /// # Examples
399 ///
400 /// ```no_run
401 /// use ibapi::client::blocking::Client;
402 ///
403 /// let client = Client::connect("127.0.0.1:4002", 100).expect("connection failed");
404 /// let codes = client.family_codes().expect("error requesting family codes");
405 /// println!("family codes: {codes:?}")
406 /// ```
407 pub fn family_codes(&self) -> Result<Vec<FamilyCode>, Error> {
408 request_helpers::blocking::one_shot_shared(
409 self,
410 OutgoingMessages::RequestFamilyCodes,
411 encoders::encode_request_family_codes,
412 expect_proto(decoders::decode_family_codes_proto),
413 )
414 .or_else(empty_on_end_of_stream)
415 }
416
417 /// Request the configured soft dollar tiers available to the account.
418 ///
419 /// # Examples
420 ///
421 /// ```no_run
422 /// use ibapi::client::blocking::Client;
423 ///
424 /// let client = Client::connect("127.0.0.1:4002", 100).expect("connection failed");
425 ///
426 /// let tiers = client.soft_dollar_tiers().expect("request failed");
427 /// for tier in &tiers {
428 /// println!("{}: {}", tier.name, tier.display_name);
429 /// }
430 /// ```
431 pub fn soft_dollar_tiers(&self) -> Result<Vec<crate::orders::SoftDollarTier>, Error> {
432 request_helpers::blocking::one_shot_by_request_id(
433 self,
434 encoders::encode_request_soft_dollar_tiers,
435 expect_proto(decoders::decode_soft_dollar_tiers_proto),
436 )
437 }
438
439 /// Request white-branding identity information for the logged-in user.
440 ///
441 /// # Examples
442 ///
443 /// ```no_run
444 /// use ibapi::client::blocking::Client;
445 ///
446 /// let client = Client::connect("127.0.0.1:4002", 100).expect("connection failed");
447 ///
448 /// let info = client.user_info().expect("request failed");
449 /// println!("white branding id: {}", info.white_branding_id);
450 /// ```
451 pub fn user_info(&self) -> Result<UserInfo, Error> {
452 request_helpers::blocking::one_shot_by_request_id(self, encoders::encode_request_user_info, expect_proto(decoders::decode_user_info_proto))
453 }
454
455 /// Request the current Financial Advisor configuration as an XML string.
456 ///
457 /// # Arguments
458 /// * `fa_data_type` - which FA dataset to fetch.
459 ///
460 /// # Errors
461 /// [`Error::Timeout`] if TWS sends no answer within 30 seconds, or within [`one_shot_timeout`](crate::client::blocking::ClientBuilder::one_shot_timeout) when set.
462 ///
463 /// # Examples
464 ///
465 /// ```no_run
466 /// use ibapi::client::blocking::Client;
467 /// use ibapi::accounts::FaDataType;
468 ///
469 /// let client = Client::connect("127.0.0.1:4002", 100).expect("connection failed");
470 ///
471 /// let cfg = client.request_fa(FaDataType::Groups).expect("request failed");
472 /// println!("{}", cfg.xml);
473 /// ```
474 pub fn request_fa(&self, fa_data_type: FaDataType) -> Result<FaConfig, Error> {
475 request_helpers::blocking::one_shot_shared_matching(
476 self,
477 OutgoingMessages::RequestFA,
478 || encoders::encode_request_fa(fa_data_type),
479 decoders::receive_fa_for(fa_data_type),
480 )
481 }
482
483 /// Replace the Financial Advisor configuration on the server.
484 ///
485 /// # Arguments
486 /// * `fa_data_type` - which FA dataset to replace.
487 /// * `xml` - the replacement configuration as an XML string.
488 ///
489 /// # Examples
490 ///
491 /// ```no_run
492 /// use ibapi::client::blocking::Client;
493 /// use ibapi::accounts::FaDataType;
494 ///
495 /// let client = Client::connect("127.0.0.1:4002", 100).expect("connection failed");
496 ///
497 /// let result = client.replace_fa(FaDataType::Groups, "<xml/>").expect("request failed");
498 /// println!("{}", result.text);
499 /// ```
500 pub fn replace_fa(&self, fa_data_type: FaDataType, xml: &str) -> Result<ReplaceFaResult, Error> {
501 request_helpers::blocking::one_shot_by_request_id(
502 self,
503 |request_id| encoders::encode_replace_fa(request_id, fa_data_type, xml),
504 expect_proto(decoders::decode_replace_fa_end_proto),
505 )
506 }
507
508 /// Set the verbosity level for server-side TWS API diagnostics.
509 ///
510 /// # Examples
511 ///
512 /// ```no_run
513 /// use ibapi::client::blocking::Client;
514 /// use ibapi::accounts::ServerLogLevel;
515 ///
516 /// let client = Client::connect("127.0.0.1:4002", 100).expect("connection failed");
517 ///
518 /// client.set_server_log_level(ServerLogLevel::Detail).expect("request failed");
519 /// ```
520 pub fn set_server_log_level(&self, log_level: ServerLogLevel) -> Result<(), Error> {
521 let message = encoders::encode_set_server_log_level(log_level)?;
522 self.send_message(message)?;
523 Ok(())
524 }
525
526 /// Initiate a TWS extension verification handshake.
527 ///
528 /// Most users will not call this directly; it is part of the IB Linking flow.
529 ///
530 /// # Errors
531 /// [`Error::Timeout`] if TWS sends no answer within 30 seconds, or within [`one_shot_timeout`](crate::client::blocking::ClientBuilder::one_shot_timeout) when set.
532 ///
533 /// # Examples
534 ///
535 /// ```no_run
536 /// use ibapi::client::blocking::Client;
537 ///
538 /// let client = Client::connect("127.0.0.1:4002", 100).expect("connection failed");
539 ///
540 /// let challenge = client.verify_request("MyApp", "1.0").expect("request failed");
541 /// println!("{}", challenge.api_data);
542 /// ```
543 pub fn verify_request(&self, api_name: &str, api_version: &str) -> Result<VerificationChallenge, Error> {
544 request_helpers::blocking::one_shot_shared(
545 self,
546 OutgoingMessages::VerifyRequest,
547 || encoders::encode_verify_request(api_name, api_version),
548 expect_proto(decoders::decode_verify_message_api_proto),
549 )
550 }
551
552 /// Continue a TWS extension verification handshake by sending the API response data.
553 ///
554 /// # Errors
555 /// [`Error::Timeout`] if TWS sends no answer within 30 seconds, or within [`one_shot_timeout`](crate::client::blocking::ClientBuilder::one_shot_timeout) when set.
556 ///
557 /// # Examples
558 ///
559 /// ```no_run
560 /// use ibapi::client::blocking::Client;
561 ///
562 /// let client = Client::connect("127.0.0.1:4002", 100).expect("connection failed");
563 ///
564 /// let result = client.verify_message("signed-challenge").expect("request failed");
565 /// if result.is_successful {
566 /// println!("verified");
567 /// } else {
568 /// eprintln!("{}", result.error_text);
569 /// }
570 /// ```
571 pub fn verify_message(&self, api_data: &str) -> Result<VerificationResult, Error> {
572 request_helpers::blocking::one_shot_shared(
573 self,
574 OutgoingMessages::VerifyMessage,
575 || encoders::encode_verify_message(api_data),
576 expect_proto(decoders::decode_verify_completed_proto),
577 )
578 }
579}
580
581/// Complete account summary snapshots from [`Client::account_summary_snapshots`].
582pub struct AccountSummarySnapshots {
583 subscription: Subscription<AccountSummaryResult>,
584 builder: SnapshotBuilder,
585 quiet: Duration,
586}
587
588impl std::fmt::Debug for AccountSummarySnapshots {
589 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
590 f.debug_struct("AccountSummarySnapshots")
591 .field("quiet", &self.quiet)
592 .finish_non_exhaustive()
593 }
594}
595
596impl AccountSummarySnapshots {
597 /// Cancels the underlying subscription. Iteration then ends, after a final snapshot of any rows
598 /// not yet emitted.
599 pub fn cancel(&self) {
600 self.subscription.cancel();
601 }
602}
603
604impl Iterator for AccountSummarySnapshots {
605 type Item = Result<AccountSummarySnapshot, Error>;
606
607 /// Returns the next snapshot, or `None` once the subscription has ended.
608 ///
609 /// A snapshot completes at an `End` marker or after the `quiet` period without a row. Rows
610 /// received before the subscription ends are returned as a final snapshot.
611 fn next(&mut self) -> Option<Self::Item> {
612 loop {
613 match self.subscription.next_batch(self.quiet) {
614 Some(Ok(batch)) => {
615 if let Some(snapshot) = self.builder.fold(batch) {
616 return Some(Ok(snapshot));
617 }
618 }
619 Some(Err(e)) => return Some(Err(e)),
620 None => return self.builder.flush().map(Ok),
621 }
622 }
623 }
624}
625
626#[cfg(test)]
627#[path = "sync_tests.rs"]
628mod tests;