### Question:
DefaultPriceService implements PriceService { Flux<Map<String, Object>> selectOnlyPriceUpdateEvents(Flux<Map<String, Object>> input) { return Flux.never(); } DefaultPriceService(CryptoService cryptoService); Flux<MessageDTO<Float>> pricesStream(Flux<Long> intervalPreferencesStream);  }### Answer:
@Test public void verifyBuildingCurrentPriceEvents() { StepVerifier.create( new DefaultPriceService(cryptoService).selectOnlyPriceUpdateEvents( Flux.just( map().put("Invalid", "A").build(), map().put(TYPE_KEY, "1").build(), map().put(TYPE_KEY, "5").build(), map().put(TYPE_KEY, "5") .put(PRICE_KEY, 0.1F) .put(CURRENCY_KEY, "USD") .put(MARKET_KEY, "External").build() ) ) ) .expectNext( map().put(TYPE_KEY, "5") .put(PRICE_KEY, 0.1F) .put(CURRENCY_KEY, "USD") .put(MARKET_KEY, "External").build() ) .expectComplete() .verify(Duration.ofSeconds(2)); }