本文整理了Java中io.reactivex.Observable.takeLast()方法的一些代码示例，展示了Observable.takeLast()的具体用法。这些代码示例主要来源于Github/Stackoverflow/Maven等平台，是从一些精选项目中提取出来的代码，具有较强的参考意义，能在一定程度帮忙到你。Observable.takeLast()方法的具体详情如下：
包路径：io.reactivex.Observable
类名称：Observable
方法名：takeLast

Observable.takeLast介绍

[英]Returns an Observable that emits at most the last count items emitted by the source ObservableSource. If the source emits fewer than count items then all of its items are emitted.

Scheduler: This version of takeLast does not operate by default on a particular Scheduler.
[中]返回最多发射源ObservableSource发射的最后计数项的Observable。如果源发射的项目少于计数，则发射其所有项目。
调度程序：默认情况下，此版本的takeLast不会在特定调度程序上运行。

代码示例

代码示例来源：origin: ReactiveX/RxJava

@Override
  public ObservableSource<Object> apply(Observable<Object> f) throws Exception {
    return f.takeLast(1);
  }
});

代码示例来源：origin: ReactiveX/RxJava

@Override
  public ObservableSource<Object> apply(Observable<Object> o) throws Exception {
    return o.takeLast(5);
  }
});

代码示例来源：origin: ReactiveX/RxJava

@Test
public void testTakeLastWithZeroCount() {
  Observable<String> w = Observable.just("one");
  Observable<String> take = w.takeLast(0);
  Observer<String> observer = TestHelper.mockObserver();
  take.subscribe(observer);
  verify(observer, never()).onNext("one");
  verify(observer, never()).onError(any(Throwable.class));
  verify(observer, times(1)).onComplete();
}

代码示例来源：origin: ReactiveX/RxJava

@Test
public void testTakeLast2() {
  Observable<String> w = Observable.just("one");
  Observable<String> take = w.takeLast(10);
  Observer<String> observer = TestHelper.mockObserver();
  take.subscribe(observer);
  verify(observer, times(1)).onNext("one");
  verify(observer, never()).onError(any(Throwable.class));
  verify(observer, times(1)).onComplete();
}

代码示例来源：origin: ReactiveX/RxJava

@Override
  public Observable<Observable<Object>> apply(Observable<Object> f)
      throws Exception {
    return f.window(Observable.never()).takeLast(1);
  }
});

代码示例来源：origin: ReactiveX/RxJava

@Test
public void takeLastTime() {
  Observable.just(1, 2)
  .takeLast(1, TimeUnit.MINUTES)
  .test()
  .assertResult(1, 2);
}

代码示例来源：origin: ReactiveX/RxJava

@Test
public void takeLastTimeAndSize() {
  Observable.just(1, 2)
  .takeLast(1, 1, TimeUnit.MINUTES)
  .test()
  .assertResult(2);
}

代码示例来源：origin: ReactiveX/RxJava

@Test
  public void testLastOfOneReturnsLast() {
    TestObserver<Integer> to = new TestObserver<Integer>();
    Observable.just(1).takeLast(1).subscribe(to);
    to.assertValue(1);
    to.assertNoErrors();
    to.assertTerminated();
    // NO longer assertable
//      s.assertUnsubscribed();
  }

代码示例来源：origin: ReactiveX/RxJava

@Test
public void takeLastTimeDelayErrorCustomScheduler() {
  Observable.just(1, 2).concatWith(Observable.<Integer>error(new TestException()))
  .takeLast(1, TimeUnit.MINUTES, Schedulers.io(), true)
  .test()
  .assertFailure(TestException.class, 1, 2);
}

代码示例来源：origin: ReactiveX/RxJava

@Test
public void testBackpressure1() {
  TestObserver<Integer> to = new TestObserver<Integer>();
  Observable.range(1, 100000).takeLast(1)
  .observeOn(Schedulers.newThread())
  .map(newSlowProcessor()).subscribe(to);
  to.awaitTerminalEvent();
  to.assertNoErrors();
  to.assertValue(100000);
}

代码示例来源：origin: ReactiveX/RxJava

@Test
public void fromArityArgs1() {
  Observable<String> items = Observable.just("one");
  assertEquals((Long)1L, items.count().blockingGet());
  assertEquals("one", items.takeLast(1).blockingSingle());
}

代码示例来源：origin: ReactiveX/RxJava

@Test
public void takeLastTimeDelayError() {
  Observable.just(1, 2).concatWith(Observable.<Integer>error(new TestException()))
  .takeLast(1, TimeUnit.MINUTES, true)
  .test()
  .assertFailure(TestException.class, 1, 2);
}

代码示例来源：origin: ReactiveX/RxJava

@Test
  public void testLastOfManyReturnsLast() {
    TestObserver<Integer> to = new TestObserver<Integer>();
    Observable.range(1, 10).takeLast(1).subscribe(to);
    to.assertValue(10);
    to.assertNoErrors();
    to.assertTerminated();
    // NO longer assertable
//        s.assertUnsubscribed();
  }

代码示例来源：origin: ReactiveX/RxJava

@Test
public void error() {
  Observable.error(new TestException())
  .takeLast(5)
  .test()
  .assertFailure(TestException.class);
}

代码示例来源：origin: ReactiveX/RxJava

@Test
public void fromIterable() {
  ArrayList<String> items = new ArrayList<String>();
  items.add("one");
  items.add("two");
  items.add("three");
  assertEquals((Long)3L, Observable.fromIterable(items).count().blockingGet());
  assertEquals("two", Observable.fromIterable(items).skip(1).take(1).blockingSingle());
  assertEquals("three", Observable.fromIterable(items).takeLast(1).blockingSingle());
}

代码示例来源：origin: ReactiveX/RxJava

@Test
  public void error() {
    Observable.error(new TestException())
    .takeLast(1)
    .test()
    .assertFailure(TestException.class);
  }
}

代码示例来源：origin: ReactiveX/RxJava

@Test
public void testBackpressure2() {
  TestObserver<Integer> to = new TestObserver<Integer>();
  Observable.range(1, 100000).takeLast(Flowable.bufferSize() * 4)
  .observeOn(Schedulers.newThread()).map(newSlowProcessor()).subscribe(to);
  to.awaitTerminalEvent();
  to.assertNoErrors();
  assertEquals(Flowable.bufferSize() * 4, to.valueCount());
}

代码示例来源：origin: ReactiveX/RxJava

@Test
  public void takeLastTake() {
    Observable.range(1, 10)
    .takeLast(5)
    .take(2)
    .test()
    .assertResult(6, 7);
  }
}

代码示例来源：origin: ReactiveX/RxJava

@Test
public void fromArityArgs3() {
  Observable<String> items = Observable.just("one", "two", "three");
  assertEquals((Long)3L, items.count().blockingGet());
  assertEquals("two", items.skip(1).take(1).blockingSingle());
  assertEquals("three", items.takeLast(1).blockingSingle());
}

代码示例来源：origin: ReactiveX/RxJava

@Test
public void fromArray() {
  String[] items = new String[] { "one", "two", "three" };
  assertEquals((Long)3L, Observable.fromArray(items).count().blockingGet());
  assertEquals("two", Observable.fromArray(items).skip(1).take(1).blockingSingle());
  assertEquals("three", Observable.fromArray(items).takeLast(1).blockingSingle());
}