OperationElementAt { public static <T> OnSubscribeFunc<T> elementAt(Observable<? extends T> source, int index) { return new ElementAt<T>(source, index, null, false); }  static OnSubscribeFunc<T> elementAt(Observable<? extends T> source, int index); static OnSubscribeFunc<T> elementAtOrDefault(Observable<? extends T> source, int index, T defaultValue);  }
@Test public void testElementAt() { Observable<Integer> w = Observable.from(1, 2); Observable<Integer> observable = Observable.create(elementAt(w, 1)); @SuppressWarnings("unchecked") Observer<Integer> observer = mock(Observer.class); observable.subscribe(observer); verify(observer, never()).onNext(1); verify(observer, times(1)).onNext(2); verify(observer, never()).onError( any(Throwable.class)); verify(observer, times(1)).onCompleted(); }
@Test public void testElementAtWithMinusIndex() { Observable<Integer> w = Observable.from(1, 2); Observable<Integer> observable = Observable .create(elementAt(w, -1)); try { Iterator<Integer> iter = OperationToIterator .toIterator(observable); assertTrue(iter.hasNext()); iter.next(); fail("expect an IndexOutOfBoundsException when index is out of bounds"); } catch (IndexOutOfBoundsException e) { } }
@Test public void testElementAtWithIndexOutOfBounds() throws InterruptedException, ExecutionException { Observable<Integer> w = Observable.from(1, 2); Observable<Integer> observable = Observable.create(elementAt(w, 2)); try { Iterator<Integer> iter = OperationToIterator .toIterator(observable); assertTrue(iter.hasNext()); iter.next(); fail("expect an IndexOutOfBoundsException when index is out of bounds"); } catch (IndexOutOfBoundsException e) { } }