StatisticsController { @GetMapping(value = "/statistics/article", produces = TEXT_EVENT_STREAM_VALUE) public Flux<ArticleStatistics> fetchArticleStatistics(@RequestParam(required = false, defaultValue = "1000") int updateInterval) { return articleReadEventsService.readEvents() .onBackpressureDrop(articleReadEvent -> logger.warn("dropping articleReadEvent: " + articleReadEvent)) .bufferMillis(updateInterval / 3) .filter(articleReadEvents -> !articleReadEvents.isEmpty()) .flatMap(articleReadEvent -> { Flux<ArticleName> distinctArticleNames = Flux.fromIterable(articleReadEvent) .map(ArticleReadEvent::toArticleName) .distinct() .cache(); Flux<Map<String, Integer>> rating = articleMetricsService.fetchRatings(distinctArticleNames) .collectMap(Rating::getArticleName, Rating::getRating) .cache() .repeat(articleReadEvent.size()); Flux<Map<String, Integer>> wordCount = articleMetricsService.fetchWordCounts(distinctArticleNames) .collectMap(WordCount::getArticleName, WordCount::getCount) .cache() .repeat(articleReadEvent.size()); return Flux.zip(Flux.fromIterable(articleReadEvent), rating, wordCount) .map(zip -> new ArticleMetrics(zip.getT2().get(zip.getT1().getArticleName()), zip.getT3().get(zip.getT1().getArticleName()), zip.getT1().getFetchTimeInMillis())); }) .bufferMillis(updateInterval) .map(StatisticsController::calculateArticleStatistics) .retry(throwable -> { logger.warn("error on fetchArticleStatistics, retrying", throwable); return true; }) .onBackpressureDrop(articleStatistics -> logger.warn("dropping articleStatistics: " + articleStatistics)) .log(); } @Autowired  StatisticsController(ArticleReadEventsService articleReadEventsService, ArticleMetricsService articleMetricsService); @GetMapping(value = "/top/article", produces = TEXT_EVENT_STREAM_VALUE) Flux<List<TopArticle>> topArticle(@RequestParam(required = false, defaultValue = "1000") int updateInterval,
                                             @RequestParam(required = false, defaultValue = "5") int numberOfTopArticles); @GetMapping(value = "/statistics/article", produces = TEXT_EVENT_STREAM_VALUE) Flux<ArticleStatistics> fetchArticleStatistics(@RequestParam(required = false, defaultValue = "1000") int updateInterval);  }
@Test public void fetchArticleStatisticsWithShortUpdateInterval() { when(articleReadEventsServiceMock.readEvents()).thenReturn(Flux.intervalMillis(310).take(4) .map(count -> createReadEvent(count, count.intValue()))); when(articleMetricsServiceMock.fetchRatings(any())).thenAnswer(invocation -> { Flux<ArticleName> names = invocation.getArgument(0); System.out.println("fetchRatings: " + names.collectList().block()); return names.map(name -> new Rating(name.getName(), Integer.parseInt(name.getName()))); }); when(articleMetricsServiceMock.fetchWordCounts(any())).thenAnswer(invocation -> { Flux<ArticleName> names = invocation.getArgument(0); return names.map(name -> new WordCount(name.getName(), Integer.parseInt(name.getName()) * 100)); }); Flux<ArticleStatistics> result = statisticsController.fetchArticleStatistics(300) .doOnNext(articleStatistics -> System.out.println("received on clientside: " + articleStatistics)); StepVerifier.create(result) .expectSubscription() .expectNext(new ArticleStatistics(1, 0.0, 0.0, 0.0)) .expectNext(new ArticleStatistics(1, 100.0, 1.0, 1.0)) .expectNext(new ArticleStatistics(1, 200.0, 2.0, 2.0)) .expectNext(new ArticleStatistics(1, 300.0, 3.0, 3.0)) .thenCancel() .verify(); }