RxJava: process imperative task using reactive approach

Question

Here is the following classes structure:

class SomeEntity {
    private Collection<SomeData> dataCollection;
    private BigDecimal price;
    ...
}

class SomeData {
    private Long personId;
    ...
}

class Person {
    private Collection<AddressLink> addressesLinks; //AddressLink is a representation of the link to the address entity
    ...
} 

I need to create ShipmentData per address. It can be implemented in such a way using imperative approach:

var loadedPersons = entity.getDataCollection().stream()
                .map(SomeData::getPersonId)
                .map(personRepository::loadById)
                .collect(Collectors.toList());

var personToAddressesLinkMap = loadedPersons.stream()
        .collect(Collectors.toMap(key -> key, val -> List.of(val.getAddressesLinks())));

var result = createShipmentData(entity, personToAddressesLinkMap);

And ShipmentData creation method could look like this:

private static List<ShipmentData> createShipmentData(SomeEntity entity, Map<Person, List<Collection<AddressLink>>> personToAddressesLinkMap) {
    List<ShipmentData> result = new ArrayList<>();

    ...

    ShipmentData data = new ShipmentData();
    data.setTargetId(...); // Person.addressLink
    data.setRecepientId(...); // SomeEntity.someData.personId
    data.setPrice(...); // SomeEntity.price
    data.setSourceId(...); //SomeEntity.someData.id
    data.setCreationDate(LocalDateTime.now());

    ...

    return result;
}

But I ran at a similar task that should be done using reactive approach (RxJava). But I cannot start thinking in terms of reactive paradigm. Here is what I have:

Observable.fromIterable(entity.getDataCollection())
    .flatMapSingle(someData -> personRepository.loadById(someData.getPersonId()))
    .flatMapIterable(person -> person.getAddressesLinks())
    .flatMapSingle(addressLink -> addressRepository.findAddressByLink(addressLink))
    .map(address -> createShipmentData(address))
    ...
    
private createShipmentData(Address address) {
    ShipmentData data = new ShipmentData();
    
    data.setTargetId(...); // Person.addressLink
    data.setRecepientId(...); // SomeEntity.someData.personId
    data.setPrice(...); // SomeEntity.price
    data.setSourceId(...); //SomeEntity.someData.id
    data.setCreationDate(LocalDateTime.now());

    return data;
}

This reactive stream narrowed the data to Address object in terms of access. And I don't have the access neither to person not to addressLink, which I need during ShipmentData construction.

How is it expected to process such scenarios using reactive approach? I was thinking about collecting all the data in some external variables and then pass all that data to createShipmentData method. But it sounds to me like a code smell.

Answer 1

The simplest way is to using nesting to keep access to the variables you need to reference. The following chain gives you a Obsevable<ShipmentData>:

Observable.fromIterable(entity.getDataCollection())
    .flatMap(someData -> personRepository.loadById(someData.getPersonId())
        .flatMapObservable(person -> Observable.fromIterable(person.getAddressesLinks())
            .flatMapSingle(addressLink -> addressRepository.findAddressByLink(addressLink))
            .map(address -> createShipmentData(someData, person, address))
        )
    )

Adding a helper method cleans it up some:

private Observable<Address> getAddresses(Person person) {
  return Observable.fromIterable(person.getAddressesLinks())
      .flatMapSingle(addressLink -> addressRepository.findAddressByLink(addressLink));
}

Observable.fromIterable(entity.getDataCollection())
    .flatMap(someData -> personRepository.loadById(someData.getPersonId())
        .flatMapObservable(person -> getAddresses(person)
            .map(address -> createShipmentData(someData, person, address))
        )
    )

You can also use some techniques to use container classes like Tuples or Pairs to pass all the arguments you need down the chain, but I think it is usually not worth the effort.