### Question:
Puller { public Publisher<Item> list(int count) { Publisher<Item> source = dbClient.getStreamOfItems(); TakeFilterOperator<Item> takeFilter = new TakeFilterOperator<>(source, count, this::isValid); return takeFilter; }  Publisher<Item> list(int count);  }### Answer:
@Test public void pullTest() throws InterruptedException { CountDownLatch l = new CountDownLatch(1); Puller puller = new Puller(); puller.list(10) .subscribe(new Subscriber<Item>() { final ArrayList<Item> list = new ArrayList<>(); @Override public void onSubscribe(Subscription s) { s.request(Long.MAX_VALUE); } @Override public void onNext(Item item) { list.add(item); } @Override public void onError(Throwable t) { } @Override public void onComplete() { MatcherAssert.assertThat(list, Matchers.allOf( Matchers.hasSize(10), Matchers.contains( Matchers.hasProperty("id", Matchers.equalTo("2")), Matchers.hasProperty("id", Matchers.equalTo("4")), Matchers.hasProperty("id", Matchers.equalTo("6")), Matchers.hasProperty("id", Matchers.equalTo("8")), Matchers.hasProperty("id", Matchers.equalTo("10")), Matchers.hasProperty("id", Matchers.equalTo("12")), Matchers.hasProperty("id", Matchers.equalTo("14")), Matchers.hasProperty("id", Matchers.equalTo("16")), Matchers.hasProperty("id", Matchers.equalTo("18")), Matchers.hasProperty("id", Matchers.equalTo("20")) ) )); l.countDown(); } }); if(!l.await(2, TimeUnit.SECONDS)) { throw new RuntimeException(); } }