OperationTakeLast { public static <T> OnSubscribeFunc<T> takeLast(final Observable<? extends T> items, final int count) { return new OnSubscribeFunc<T>() { @Override public Subscription onSubscribe(Observer<? super T> observer) { return new TakeLast<T>(items, count).onSubscribe(observer); } }; }  static OnSubscribeFunc<T> takeLast(final Observable<? extends T> items, final int count); static OnSubscribeFunc<T> takeLast(Observable<? extends T> source, long time, TimeUnit unit, Scheduler scheduler); static OnSubscribeFunc<T> takeLast(Observable<? extends T> source, int count, long time, TimeUnit unit, Scheduler scheduler);  }
@Test public void testTakeLastEmpty() { Observable<String> w = Observable.empty(); Observable<String> take = Observable.create(takeLast(w, 2)); @SuppressWarnings("unchecked") Observer<String> observer = mock(Observer.class); take.subscribe(observer); verify(observer, never()).onNext(any(String.class)); verify(observer, never()).onError(any(Throwable.class)); verify(observer, times(1)).onCompleted(); }
@Test public void testTakeLast1() { Observable<String> w = Observable.from("one", "two", "three"); Observable<String> take = Observable.create(takeLast(w, 2)); @SuppressWarnings("unchecked") Observer<String> observer = mock(Observer.class); InOrder inOrder = inOrder(observer); take.subscribe(observer); inOrder.verify(observer, times(1)).onNext("two"); inOrder.verify(observer, times(1)).onNext("three"); verify(observer, never()).onNext("one"); verify(observer, never()).onError(any(Throwable.class)); verify(observer, times(1)).onCompleted(); }
@Test public void testTakeLast2() { Observable<String> w = Observable.from("one"); Observable<String> take = Observable.create(takeLast(w, 10)); @SuppressWarnings("unchecked") Observer<String> observer = mock(Observer.class); take.subscribe(observer); verify(observer, times(1)).onNext("one"); verify(observer, never()).onError(any(Throwable.class)); verify(observer, times(1)).onCompleted(); }
@Test public void testTakeLastWithZeroCount() { Observable<String> w = Observable.from("one"); Observable<String> take = Observable.create(takeLast(w, 0)); @SuppressWarnings("unchecked") Observer<String> observer = mock(Observer.class); take.subscribe(observer); verify(observer, never()).onNext("one"); verify(observer, never()).onError(any(Throwable.class)); verify(observer, times(1)).onCompleted(); }
@Test public void testTakeLastWithNull() { Observable<String> w = Observable.from("one", null, "three"); Observable<String> take = Observable.create(takeLast(w, 2)); @SuppressWarnings("unchecked") Observer<String> observer = mock(Observer.class); take.subscribe(observer); verify(observer, never()).onNext("one"); verify(observer, times(1)).onNext(null); verify(observer, times(1)).onNext("three"); verify(observer, never()).onError(any(Throwable.class)); verify(observer, times(1)).onCompleted(); }
@Test public void testTakeLastWithNegativeCount() { Observable<String> w = Observable.from("one"); Observable<String> take = Observable.create(takeLast(w, -1)); @SuppressWarnings("unchecked") Observer<String> observer = mock(Observer.class); take.subscribe(observer); verify(observer, never()).onNext("one"); verify(observer, times(1)).onError( any(IndexOutOfBoundsException.class)); verify(observer, never()).onCompleted(); }
@Test public void takeLastTimed() { TestScheduler scheduler = new TestScheduler(); PublishSubject<Object> source = PublishSubject.create(); Observable<Object> result = source.takeLast(1, TimeUnit.SECONDS, scheduler); Observer<Object> o = mock(Observer.class); InOrder inOrder = inOrder(o); result.subscribe(o); source.onNext(1); scheduler.advanceTimeBy(250, TimeUnit.MILLISECONDS); source.onNext(2); scheduler.advanceTimeBy(250, TimeUnit.MILLISECONDS); source.onNext(3); scheduler.advanceTimeBy(250, TimeUnit.MILLISECONDS); source.onNext(4); scheduler.advanceTimeBy(250, TimeUnit.MILLISECONDS); source.onNext(5); scheduler.advanceTimeBy(250, TimeUnit.MILLISECONDS); source.onCompleted(); inOrder.verify(o, times(1)).onNext(2); inOrder.verify(o, times(1)).onNext(3); inOrder.verify(o, times(1)).onNext(4); inOrder.verify(o, times(1)).onNext(5); inOrder.verify(o, times(1)).onCompleted(); verify(o, never()).onError(any(Throwable.class)); }
@Test public void takeLastTimedDelayCompletion() { TestScheduler scheduler = new TestScheduler(); PublishSubject<Object> source = PublishSubject.create(); Observable<Object> result = source.takeLast(1, TimeUnit.SECONDS, scheduler); Observer<Object> o = mock(Observer.class); InOrder inOrder = inOrder(o); result.subscribe(o); source.onNext(1); scheduler.advanceTimeBy(250, TimeUnit.MILLISECONDS); source.onNext(2); scheduler.advanceTimeBy(250, TimeUnit.MILLISECONDS); source.onNext(3); scheduler.advanceTimeBy(250, TimeUnit.MILLISECONDS); source.onNext(4); scheduler.advanceTimeBy(250, TimeUnit.MILLISECONDS); source.onNext(5); scheduler.advanceTimeBy(1250, TimeUnit.MILLISECONDS); source.onCompleted(); inOrder.verify(o, times(1)).onCompleted(); verify(o, never()).onNext(any()); verify(o, never()).onError(any(Throwable.class)); }
@Test public void takeLastTimedWithCapacity() { TestScheduler scheduler = new TestScheduler(); PublishSubject<Object> source = PublishSubject.create(); Observable<Object> result = source.takeLast(2, 1, TimeUnit.SECONDS, scheduler); Observer<Object> o = mock(Observer.class); InOrder inOrder = inOrder(o); result.subscribe(o); source.onNext(1); scheduler.advanceTimeBy(250, TimeUnit.MILLISECONDS); source.onNext(2); scheduler.advanceTimeBy(250, TimeUnit.MILLISECONDS); source.onNext(3); scheduler.advanceTimeBy(250, TimeUnit.MILLISECONDS); source.onNext(4); scheduler.advanceTimeBy(250, TimeUnit.MILLISECONDS); source.onNext(5); scheduler.advanceTimeBy(250, TimeUnit.MILLISECONDS); source.onCompleted(); inOrder.verify(o, times(1)).onNext(4); inOrder.verify(o, times(1)).onNext(5); inOrder.verify(o, times(1)).onCompleted(); verify(o, never()).onError(any(Throwable.class)); }
@Test public void takeLastTimedThrowingSource() { TestScheduler scheduler = new TestScheduler(); PublishSubject<Object> source = PublishSubject.create(); Observable<Object> result = source.takeLast(1, TimeUnit.SECONDS, scheduler); Observer<Object> o = mock(Observer.class); InOrder inOrder = inOrder(o); result.subscribe(o); source.onNext(1); scheduler.advanceTimeBy(250, TimeUnit.MILLISECONDS); source.onNext(2); scheduler.advanceTimeBy(250, TimeUnit.MILLISECONDS); source.onNext(3); scheduler.advanceTimeBy(250, TimeUnit.MILLISECONDS); source.onNext(4); scheduler.advanceTimeBy(250, TimeUnit.MILLISECONDS); source.onNext(5); scheduler.advanceTimeBy(250, TimeUnit.MILLISECONDS); source.onError(new CustomException()); inOrder.verify(o, times(1)).onError(any(CustomException.class)); verify(o, never()).onNext(any()); verify(o, never()).onCompleted(); }
@Test public void takeLastTimedWithZeroCapacity() { TestScheduler scheduler = new TestScheduler(); PublishSubject<Object> source = PublishSubject.create(); Observable<Object> result = source.takeLast(0, 1, TimeUnit.SECONDS, scheduler); Observer<Object> o = mock(Observer.class); InOrder inOrder = inOrder(o); result.subscribe(o); source.onNext(1); scheduler.advanceTimeBy(250, TimeUnit.MILLISECONDS); source.onNext(2); scheduler.advanceTimeBy(250, TimeUnit.MILLISECONDS); source.onNext(3); scheduler.advanceTimeBy(250, TimeUnit.MILLISECONDS); source.onNext(4); scheduler.advanceTimeBy(250, TimeUnit.MILLISECONDS); source.onNext(5); scheduler.advanceTimeBy(250, TimeUnit.MILLISECONDS); source.onCompleted(); inOrder.verify(o, times(1)).onCompleted(); verify(o, never()).onNext(any()); verify(o, never()).onError(any(Throwable.class)); }