### Question:
DefaultTradeService implements TradeService { Flux<MessageDTO<MessageDTO.Trade>> filterAndMapTradingEvents(Flux<Map<String, Object>> input) { return Flux.never(); } DefaultTradeService(CryptoService service, TradeRepository repository); @Override Flux<MessageDTO<MessageDTO.Trade>> tradesStream();  }### Answer:
@Test public void verifyTradeInfoMappingToDTO() { StepVerifier.create( new DefaultTradeService(cryptoService, tradeRepository).filterAndMapTradingEvents( Flux.just( map().put("Invalid", "A").build(), map().put(TYPE_KEY, "1").build(), map().put(TYPE_KEY, "0") .put(TIMESTAMP_KEY, 1F) .put(PRICE_KEY, 0.1F) .put(FLAGS_KEY, "1") .put(QUANTITY_KEY, 1.2F) .put(CURRENCY_KEY, "USD") .put(MARKET_KEY, "External").build() ) ) .take(Duration.ofSeconds(1)) ) .expectNext(MessageDTO.trade(1000, 0.1F, 1.2F, "USD", "External")) .verifyComplete(); }