rxrust 1.0.0-rc.4

A Rust implementation of Reactive Extensions.
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52

//! SourceWithDynamicSubs subscription combinator
//!
//! A generic subscription that combines a source subscription with
//! a DynamicSubscriptions container. Used by operators like ObserveOn,
//! MergeAll, and Delay that manage multiple dynamic subscriptions.

use super::{DynamicSubscriptions, Subscription};
use crate::context::RcDerefMut;

/// A subscription combining a source subscription with dynamic subscriptions.
///
/// When unsubscribed, all dynamic subscriptions are drained and unsubscribed
/// first, then the source subscription is unsubscribed.
///
/// # Type Parameters
///
/// - `U`: The source subscription type
/// - `S`: The RcDerefMut container for DynamicSubscriptions
///
/// # Examples
///
/// ```ignore
/// type ObserveOnSubscription<U, S> = SourceWithDynamicSubs<U, S>;
/// type MergeAllSubscription<U, S> = SourceWithDynamicSubs<U, S>;
/// type DelaySubscription<U, S> = SourceWithDynamicSubs<U, S>;
/// ```
pub struct SourceWithDynamicSubs<U, S> {
  pub source: U,
  pub subs: S,
}

impl<U, S> SourceWithDynamicSubs<U, S> {
  /// Create a new SourceWithDynamicSubs subscription
  #[inline]
  pub fn new(source: U, subs: S) -> Self { Self { source, subs } }
}

impl<U, S, H> Subscription for SourceWithDynamicSubs<U, S>
where
  U: Subscription,
  S: RcDerefMut<Target = DynamicSubscriptions<H>>,
  H: Subscription,
{
  fn unsubscribe(self) {
    // Unsubscribe source first
    self.source.unsubscribe();
    // Then drain and unsubscribe all dynamic subs
    self.subs.rc_deref_mut().unsubscribe_all();
  }

  fn is_closed(&self) -> bool { self.source.is_closed() && self.subs.rc_deref().all_closed() }
}