### Question:
DefaultPriceService implements PriceService { Flux<MessageDTO<Float>> averagePrice( Flux<Long> requestedInterval, Flux<MessageDTO<Float>> priceData ) { return Flux.never(); } DefaultPriceService(CryptoService cryptoService); Flux<MessageDTO<Float>> pricesStream(Flux<Long> intervalPreferencesStream);  }### Answer:
@Test @SuppressWarnings("unchecked") public void verifyBuildingAveragePriceEvents() { StepVerifier.withVirtualTime(() -> new DefaultPriceService(cryptoService).averagePrice( Flux.interval(Duration.ofSeconds(0), Duration.ofSeconds(5)) .map(i -> i + 1) .doOnNext(i -> System.out.println("Interval: " + i)), Flux.interval(Duration.ofMillis(500), Duration.ofSeconds(1)) .map(p -> p + 100) .map(tick -> MessageDTO.price((float) tick, "U", "M")) .take(20) .doOnNext(p -> System.out.println("Price: " + p.getData())) .replay(1000) .autoConnect() ) .take(10) .take(Duration.ofHours(1)) .map(MessageDTO::getData) .doOnNext(a -> System.out.println("AVG: " + a)) ) .expectSubscription() .thenAwait(Duration.ofDays(1)) .expectNextMatches(expectedPrice(100.0F)) .expectNextMatches(expectedPrice(101.0F)) .expectNextMatches(expectedPrice(102.0F)) .expectNextMatches(expectedPrice(103.0F)) .expectNextMatches(expectedPrice(104.0F)) .expectNextMatches(expectedPrice(103.0F)) .expectNextMatches(expectedPrice(107.5F)) .expectNextMatches(expectedPrice(109.5F)) .expectNextMatches(expectedPrice(106.0F)) .expectNextMatches(expectedPrice(114.0F)) .verifyComplete(); }