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

Flowable.takeLast介绍

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

Backpressure: The operator honors backpressure from downstream if the count is non-zero; ignores backpressure if the count is zero as it doesn't signal any values. Scheduler: This version of takeLast does not operate by default on a particular Scheduler.
[中]返回最多发出源发布服务器发出的最后计数项的可流动项。如果源发射的项目少于计数，则发射其所有项目。
背压：如果计数不为零，操作员接受来自下游的背压；如果计数为零，则忽略背压，因为它不发出任何值的信号。调度程序：默认情况下，此版本的takeLast不会在特定调度程序上运行。

代码示例

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

@Override
  public Flowable<Object> apply(Flowable<Object> f) throws Exception {
    return f.takeLast(5);
  }
});

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

@Override
  public Publisher<Flowable<Object>> apply(Flowable<Object> f)
      throws Exception {
    return f.window(1, TimeUnit.SECONDS, 1).takeLast(0);
  }
});

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

@Override
  public Flowable<Integer> apply(Flowable<Integer> f) {
    return Flowable.concat(f.take(5), f.takeLast(5));
  }
}).subscribe(ts);

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

@Test(expected = NullPointerException.class)
public void takeLastSizeTimedUnitNull() {
  just1.takeLast(1, 1, null, Schedulers.single());
}

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

@Override
  public Flowable<Integer> apply(Flowable<Integer> f) {
    return Flowable.concat(f.take(5), f.takeLast(5));
  }
}).subscribe(ts);

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

@Override
  public Flowable<Integer> apply(Flowable<Integer> f) {
    return Flowable.concat(f.take(5), f.takeLast(5));
  }
}).subscribe(ts);

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

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

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

@Test
public void testTakeLastEmpty() {
  Flowable<String> w = Flowable.empty();
  Flowable<String> take = w.takeLast(2);
  Subscriber<String> subscriber = TestHelper.mockSubscriber();
  take.subscribe(subscriber);
  verify(subscriber, never()).onNext(any(String.class));
  verify(subscriber, never()).onError(any(Throwable.class));
  verify(subscriber, times(1)).onComplete();
}

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

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

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

@Test
public void testLastWithBackpressure() {
  MySubscriber<Integer> s = new MySubscriber<Integer>(0);
  Flowable.just(1).takeLast(1).subscribe(s);
  assertEquals(0, s.list.size());
  s.requestMore(1);
  assertEquals(1, s.list.size());
}

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

@Test
  public void testLastOfEmptyReturnsEmpty() {
    TestSubscriber<Object> s = new TestSubscriber<Object>();
    Flowable.empty().takeLast(1).subscribe(s);
    s.assertNoValues();
    s.assertNoErrors();
    s.assertTerminated();
    // NO longer assertable
//      s.assertUnsubscribed();
  }

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

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

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

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