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
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125

use crate::{impl_helper::*, impl_local_shared_both, prelude::*};

#[derive(Clone)]
pub struct SkipOp<S> {
  pub(crate) source: S,
  pub(crate) count: u32,
}

impl<S: Observable> Observable for SkipOp<S> {
  type Item = S::Item;
  type Err = S::Err;
}

impl_local_shared_both! {
 impl<S> SkipOp<S>;
 type Unsub = @ctx::Rc<ProxySubscription<S::Unsub>>;
 macro method($self: ident, $observer: ident, $ctx: ident) {
   let subscription =$ctx::Rc::own(ProxySubscription::default());
   let u = $self.source.actual_subscribe(SkipObserver {
    observer: $observer,
    subscription: subscription.clone(),
    count: $self.count,
    hits: 0,
  });
   subscription.rc_deref_mut().proxy(u);
   subscription
 }
 where
  @ctx::local_only(S::Unsub: 'o,)
  S: @ctx::Observable
}

pub struct SkipObserver<O, S> {
  observer: O,
  subscription: S,
  count: u32,
  hits: u32,
}

impl<Item, Err, O, U> Observer for SkipObserver<O, U>
where
  O: Observer<Item = Item, Err = Err>,
  U: SubscriptionLike,
{
  type Item = Item;
  type Err = Err;
  fn next(&mut self, value: Item) {
    self.hits += 1;
    if self.hits > self.count {
      self.observer.next(value);
      if self.hits == self.count {
        self.complete();
        self.subscription.unsubscribe();
      }
    }
  }

  fn error(&mut self, err: Self::Err) { self.observer.error(err) }

  fn complete(&mut self) { self.observer.complete() }
}

#[cfg(test)]
mod test {
  use crate::prelude::*;

  #[test]
  fn base_function() {
    let mut completed = false;
    let mut next_count = 0;

    observable::from_iter(0..100)
      .skip(5)
      .subscribe_complete(|_| next_count += 1, || completed = true);

    assert_eq!(next_count, 95);
    assert!(completed);
  }

  #[test]
  fn base_empty_function() {
    let mut completed = false;
    let mut next_count = 0;

    observable::from_iter(0..100)
      .skip(101)
      .subscribe_complete(|_| next_count += 1, || completed = true);

    assert_eq!(next_count, 0);
    assert!(completed);
  }

  #[test]
  fn skip_support_fork() {
    let mut nc1 = 0;
    let mut nc2 = 0;
    {
      let skip5 = observable::from_iter(0..100).skip(5);
      let f1 = skip5.clone();
      let f2 = skip5;

      f1.skip(5).subscribe(|_| nc1 += 1);
      f2.skip(5).subscribe(|_| nc2 += 1);
    }
    assert_eq!(nc1, 90);
    assert_eq!(nc2, 90);
  }

  #[cfg(not(target_arch = "wasm32"))]
  #[test]
  fn ininto_shared() {
    observable::from_iter(0..100)
      .skip(5)
      .skip(5)
      .into_shared()
      .subscribe(|_| {});
  }

  #[test]
  fn benchmark() { do_bench(); }

  benchmark_group!(do_bench, bench);

  fn bench(b: &mut bencher::Bencher) { b.iter(base_function); }
}