@Test public void testGetSightings() { Producer producer = new Producer(null, producerProperties(), this.crawler); List<Sighting> sightings = producer.getSightings() .collectList() .block(); assertThat(sightings, hasSize(31)); }
Flux<Sighting> getSightings() { YearMonth from = producerProperties.getFromYearMonth(); YearMonth to = producerProperties.getToYearMonth(); int months = (int) from.until(to, MONTHS) + 1; int numBatches = (int) Math.ceil((double) months / BATCH_SIZE); return Flux.range(1, numBatches - 1) .zipWith(Flux.intervalMillis(producerProperties.getDelayMillis()), (i, delay) -> i) .scanWith(new TupleSupplier(true, from, to), (t, i) -> new TupleSupplier(false, t.getT1(), to).get()) .groupBy(t -> t.getT1().getYear()) .flatMap(g -> g .publishOn(Schedulers.parallel()) .flatMap(t -> crawler.getSightings(t.getT1(), t.getT2())), CONCURRENCY); }
Producer implements CommandLineRunner { Flux<Sighting> getSightings() { YearMonth from = producerProperties.getFromYearMonth(); YearMonth to = producerProperties.getToYearMonth(); int months = (int) from.until(to, MONTHS) + 1; int numBatches = (int) Math.ceil((double) months / BATCH_SIZE); return Flux.range(1, numBatches - 1) .zipWith(Flux.intervalMillis(producerProperties.getDelayMillis()), (i, delay) -> i) .scanWith(new TupleSupplier(true, from, to), (t, i) -> new TupleSupplier(false, t.getT1(), to).get()) .groupBy(t -> t.getT1().getYear()) .flatMap(g -> g .publishOn(Schedulers.parallel()) .flatMap(t -> crawler.getSightings(t.getT1(), t.getT2())), CONCURRENCY); } }
Producer implements CommandLineRunner { Flux<Sighting> getSightings() { YearMonth from = producerProperties.getFromYearMonth(); YearMonth to = producerProperties.getToYearMonth(); int months = (int) from.until(to, MONTHS) + 1; int numBatches = (int) Math.ceil((double) months / BATCH_SIZE); return Flux.range(1, numBatches - 1) .zipWith(Flux.intervalMillis(producerProperties.getDelayMillis()), (i, delay) -> i) .scanWith(new TupleSupplier(true, from, to), (t, i) -> new TupleSupplier(false, t.getT1(), to).get()) .groupBy(t -> t.getT1().getYear()) .flatMap(g -> g .publishOn(Schedulers.parallel()) .flatMap(t -> crawler.getSightings(t.getT1(), t.getT2())), CONCURRENCY); }  }
Producer implements CommandLineRunner { Flux<Sighting> getSightings() { YearMonth from = producerProperties.getFromYearMonth(); YearMonth to = producerProperties.getToYearMonth(); int months = (int) from.until(to, MONTHS) + 1; int numBatches = (int) Math.ceil((double) months / BATCH_SIZE); return Flux.range(1, numBatches - 1) .zipWith(Flux.intervalMillis(producerProperties.getDelayMillis()), (i, delay) -> i) .scanWith(new TupleSupplier(true, from, to), (t, i) -> new TupleSupplier(false, t.getT1(), to).get()) .groupBy(t -> t.getT1().getYear()) .flatMap(g -> g .publishOn(Schedulers.parallel()) .flatMap(t -> crawler.getSightings(t.getT1(), t.getT2())), CONCURRENCY); }  @Override void run(String... args); }
Producer implements CommandLineRunner { Flux<Sighting> getSightings() { YearMonth from = producerProperties.getFromYearMonth(); YearMonth to = producerProperties.getToYearMonth(); int months = (int) from.until(to, MONTHS) + 1; int numBatches = (int) Math.ceil((double) months / BATCH_SIZE); return Flux.range(1, numBatches - 1) .zipWith(Flux.intervalMillis(producerProperties.getDelayMillis()), (i, delay) -> i) .scanWith(new TupleSupplier(true, from, to), (t, i) -> new TupleSupplier(false, t.getT1(), to).get()) .groupBy(t -> t.getT1().getYear()) .flatMap(g -> g .publishOn(Schedulers.parallel()) .flatMap(t -> crawler.getSightings(t.getT1(), t.getT2())), CONCURRENCY); }  @Override void run(String... args);  }