OperationTakeTimed { public static <T> OnSubscribeFunc<T> take(final Observable<? extends T> items, final int num) { return new OnSubscribeFunc<T>() { @Override public Subscription onSubscribe(Observer<? super T> observer) { return new Take<T>(items, num).onSubscribe(observer); } }; }  static OnSubscribeFunc<T> take(final Observable<? extends T> items, final int num);  }
@Test public void testTakeTimed() { TestScheduler scheduler = new TestScheduler(); PublishSubject<Integer> source = PublishSubject.create(); Observable<Integer> result = source.take(1, TimeUnit.SECONDS, scheduler); @SuppressWarnings("unchecked") Observer<Object> o = mock(Observer.class); result.subscribe(o); source.onNext(1); source.onNext(2); source.onNext(3); scheduler.advanceTimeBy(1, TimeUnit.SECONDS); source.onNext(4); InOrder inOrder = inOrder(o); inOrder.verify(o).onNext(1); inOrder.verify(o).onNext(2); inOrder.verify(o).onNext(3); inOrder.verify(o).onCompleted(); inOrder.verifyNoMoreInteractions(); verify(o, never()).onNext(4); verify(o, never()).onError(any(Throwable.class)); }
@Test public void testTakeTimedErrorBeforeTime() { TestScheduler scheduler = new TestScheduler(); PublishSubject<Integer> source = PublishSubject.create(); Observable<Integer> result = source.take(1, TimeUnit.SECONDS, scheduler); @SuppressWarnings("unchecked") Observer<Object> o = mock(Observer.class); result.subscribe(o); source.onNext(1); source.onNext(2); source.onNext(3); source.onError(new CustomException()); scheduler.advanceTimeBy(1, TimeUnit.SECONDS); source.onNext(4); InOrder inOrder = inOrder(o); inOrder.verify(o).onNext(1); inOrder.verify(o).onNext(2); inOrder.verify(o).onNext(3); inOrder.verify(o).onError(any(CustomException.class)); inOrder.verifyNoMoreInteractions(); verify(o, never()).onCompleted(); verify(o, never()).onNext(4); }
@Test public void testTakeTimedErrorAfterTime() { TestScheduler scheduler = new TestScheduler(); PublishSubject<Integer> source = PublishSubject.create(); Observable<Integer> result = source.take(1, TimeUnit.SECONDS, scheduler); @SuppressWarnings("unchecked") Observer<Object> o = mock(Observer.class); result.subscribe(o); source.onNext(1); source.onNext(2); source.onNext(3); scheduler.advanceTimeBy(1, TimeUnit.SECONDS); source.onNext(4); source.onError(new CustomException()); InOrder inOrder = inOrder(o); inOrder.verify(o).onNext(1); inOrder.verify(o).onNext(2); inOrder.verify(o).onNext(3); inOrder.verify(o).onCompleted(); inOrder.verifyNoMoreInteractions(); verify(o, never()).onNext(4); verify(o, never()).onError(any(CustomException.class)); }