Java reactor flux not mapping according expectation

user_215311 0
Viewed 656

I have the following setup for a demo application:

  • mongodb containing 2 collections: 1 with cryptocurrencies and 1 with the exchange rates of these cryptocurrencies
  • spring webflux project to fetch realtime updates to these exchange rates using Server Sent events

I have a service which returns a Flux of List<CryptoCurrencyRateDTO> based on the currencies present in the cryptocurrency collection. I generate a random exchange rate for each of these currencies and stream them to the web client.

The service is this:

@Service
public class CryptoCurrencyRateService {
  @Autowired private CryptoCurrencyRateRepository rateRepository;
  @Autowired private CryptoCurrencyRepository currencyRepository;

  // constructor

  public Flux<List<CryptoCurrencyRateDTO>> realtimeRates() {
    return currencyRepository.findAll()
      .map(CryptoCurrency::getSymbol)
      .flatMap(rateRepository::findTopBySymbolOrderByTimestamp)
      .zipWith(
        Flux.<Long>generate(sink -> sink.next(Instant.now().toEpochMilli())),
        (rate, timestamp) -> new CryptoCurrencyRate(rate.getSymbol(), timestamp, randomRateBasedOnPrevious )
      )
      .flatMap(rateRepository::save)
      .map(rateMapper::toDto)
      .collectList()
      .delayElement(Duration.ofSeconds(5))
      .repeat();
  }
}

The CryptoCurrencyRateRepository is as follows:

@Repository
public interface CryptoCurrencyRateRepository extends ReactiveMongoRepository<CryptoCurrencyRate, String> {
    Mono<CryptoCurrencyRate> findTopBySymbolOrderByTimestamp(String symbol);
}

However after the call to .flatMap(rateRepository::findTopBySymbolOrderByTimestamp) I only get a Flux containing 1 item while I thought I would get a Flux containing the top rate for each symbol from the currencyRepository.findAll().map(CryptoCurrency::getSymbol) call because my cryptocurrency collection contains 3 currencies.

When I look in the logging I see that the call to findTopBySymbolOrderByTimestamp is executed 3 times

2018-11-16 16:04:33.626 DEBUG 3387 --- [ntLoopGroup-2-3] o.s.d.m.core.ReactiveMongoTemplate       : find using query: { "symbol" : "BTC" } fields: Document{{}} for class: class nl.reactive.charts.server.domain.CryptoCurrencyRate in collection: cryptoCurrencyRate
2018-11-16 16:04:33.627 DEBUG 3387 --- [ntLoopGroup-2-3] o.s.d.m.core.ReactiveMongoTemplate       : find using query: { "symbol" : "ETH" } fields: Document{{}} for class: class nl.reactive.charts.server.domain.CryptoCurrencyRate in collection: cryptoCurrencyRate
2018-11-16 16:04:33.629 DEBUG 3387 --- [ntLoopGroup-2-3] o.s.d.m.core.ReactiveMongoTemplate       : find using query: { "symbol" : "XRP" } fields: Document{{}} for class: class nl.reactive.charts.server.domain.CryptoCurrencyRate in collection: cryptoCurrencyRate
project-reactor spring-boot java
2 Answers

I'm not able to reproduce your issue. This is how I mimic it

public static void main(String[] args) {
    Flux<String> stringFlux = Flux.fromStream(Stream.of("a", "b", "c"));
    System.out.println(realtimeRates(stringFlux).blockFirst());
}

static Flux<List<String>> realtimeRates(Flux<String> list) {
    Flux<String> symbols = list.map(Scratch::getSymbol);
    Flux<String> topRates = symbols.flatMap(Scratch::findTopBySymbolOrderByTimestamp);
    Flux<String> zip = topRates.zipWith(
        Flux.<Long>generate(sink -> sink.next(Instant.now().toEpochMilli())),
        (rate, timestamp) -> rate + timestamp.toString());
    Mono<List<String>> listMono = zip.collectList();
    Mono<List<String>> delayElement = listMono.delayElement(Duration.ofSeconds(5));
    Flux<List<String>> repeat = delayElement.repeat();
    return repeat;
}

static Mono<String> findTopBySymbolOrderByTimestamp(String symbol) {
    return Mono.just("other-" + symbol);
}

static String getSymbol(String rate) {
    return rate.toLowerCase();
}

As you can see, you will get something like [other-a1542821666133, other-b1542821666133, other-c1542821666133].

How are you checking the flatmap result? be aware that if you do it with a blockFirst() or blockLast() methods you will get only one element since its a Flux<String> (check the topRates variable in the code above)

user_24514250

It appears that my expectations were correct. The only thing that was going wrong was that not all objects got saved in Mongo because they were missing an ID.

user_2153110
Related