OperationSkipLast { public static <T> OnSubscribeFunc<T> skipLast( Observable<? extends T> source, int count) { return new SkipLast<T>(source, count); }  static OnSubscribeFunc<T> skipLast(
            Observable<? extends T> source, int count);  }
@Test public void testSkipLastEmpty() { Observable<String> w = Observable.empty(); Observable<String> observable = Observable.create(skipLast(w, 2)); @SuppressWarnings("unchecked") Observer<String> observer = mock(Observer.class); observable.subscribe(observer); verify(observer, never()).onNext(any(String.class)); verify(observer, never()).onError(any(Throwable.class)); verify(observer, times(1)).onCompleted(); }
@Test public void testSkipLast1() { Observable<String> w = Observable.from("one", "two", "three"); Observable<String> observable = Observable.create(skipLast(w, 2)); @SuppressWarnings("unchecked") Observer<String> observer = mock(Observer.class); InOrder inOrder = inOrder(observer); observable.subscribe(observer); inOrder.verify(observer, never()).onNext("two"); inOrder.verify(observer, never()).onNext("three"); verify(observer, times(1)).onNext("one"); verify(observer, never()).onError(any(Throwable.class)); verify(observer, times(1)).onCompleted(); }
@Test public void testSkipLast2() { Observable<String> w = Observable.from("one", "two"); Observable<String> observable = Observable.create(skipLast(w, 2)); @SuppressWarnings("unchecked") Observer<String> observer = mock(Observer.class); observable.subscribe(observer); verify(observer, never()).onNext(any(String.class)); verify(observer, never()).onError(any(Throwable.class)); verify(observer, times(1)).onCompleted(); }
@Test public void testSkipLastWithZeroCount() { Observable<String> w = Observable.from("one", "two"); Observable<String> observable = Observable.create(skipLast(w, 0)); @SuppressWarnings("unchecked") Observer<String> observer = mock(Observer.class); observable.subscribe(observer); verify(observer, times(1)).onNext("one"); verify(observer, times(1)).onNext("two"); verify(observer, never()).onError(any(Throwable.class)); verify(observer, times(1)).onCompleted(); }
@Test public void testSkipLastWithNull() { Observable<String> w = Observable.from("one", null, "two"); Observable<String> observable = Observable.create(skipLast(w, 1)); @SuppressWarnings("unchecked") Observer<String> observer = mock(Observer.class); observable.subscribe(observer); verify(observer, times(1)).onNext("one"); verify(observer, times(1)).onNext(null); verify(observer, never()).onNext("two"); verify(observer, never()).onError(any(Throwable.class)); verify(observer, times(1)).onCompleted(); }
@Test public void testSkipLastWithNegativeCount() { Observable<String> w = Observable.from("one"); Observable<String> observable = Observable.create(skipLast(w, -1)); @SuppressWarnings("unchecked") Observer<String> observer = mock(Observer.class); observable.subscribe(observer); verify(observer, never()).onNext(any(String.class)); verify(observer, times(1)).onError( any(IndexOutOfBoundsException.class)); verify(observer, never()).onCompleted(); }
@Test public void testSkipLastTimed() { TestScheduler scheduler = new TestScheduler(); PublishSubject<Integer> source = PublishSubject.create(); Observable<Integer> result = source.skipLast(1, TimeUnit.SECONDS, scheduler); Observer<Object> o = mock(Observer.class); result.subscribe(o); source.onNext(1); source.onNext(2); source.onNext(3); scheduler.advanceTimeBy(500, TimeUnit.MILLISECONDS); source.onNext(4); source.onNext(5); source.onNext(6); scheduler.advanceTimeBy(950, TimeUnit.MILLISECONDS); source.onCompleted(); InOrder inOrder = inOrder(o); inOrder.verify(o).onNext(1); inOrder.verify(o).onNext(2); inOrder.verify(o).onNext(3); inOrder.verify(o, never()).onNext(4); inOrder.verify(o, never()).onNext(5); inOrder.verify(o, never()).onNext(6); inOrder.verify(o).onCompleted(); inOrder.verifyNoMoreInteractions(); verify(o, never()).onError(any(Throwable.class)); }
@Test public void testSkipLastTimedErrorBeforeTime() { TestScheduler scheduler = new TestScheduler(); PublishSubject<Integer> source = PublishSubject.create(); Observable<Integer> result = source.skipLast(1, TimeUnit.SECONDS, scheduler); Observer<Object> o = mock(Observer.class); result.subscribe(o); source.onNext(1); source.onNext(2); source.onNext(3); source.onError(new OperationSkipTest.CustomException()); scheduler.advanceTimeBy(1050, TimeUnit.MILLISECONDS); verify(o).onError(any(CustomException.class)); verify(o, never()).onCompleted(); verify(o, never()).onNext(any()); }
@Test public void testSkipLastTimedCompleteBeforeTime() { TestScheduler scheduler = new TestScheduler(); PublishSubject<Integer> source = PublishSubject.create(); Observable<Integer> result = source.skipLast(1, TimeUnit.SECONDS, scheduler); Observer<Object> o = mock(Observer.class); result.subscribe(o); source.onNext(1); source.onNext(2); source.onNext(3); scheduler.advanceTimeBy(500, TimeUnit.MILLISECONDS); source.onCompleted(); InOrder inOrder = inOrder(o); inOrder.verify(o).onCompleted(); inOrder.verifyNoMoreInteractions(); verify(o, never()).onNext(any()); verify(o, never()).onError(any(Throwable.class)); }