Modify Mono if other Mono is empty in project Reactor?

Question

I'm having a Mono that I want to modify when other Mono is empty.

First Mono:

deviceRepository.getDevice(deviceId, userId): Mono<Device>

Second Mono:

registrationRepository.canRegister(deviceId) : Mono<DeviceRegistrationInfo>

What I want to achieve is return modified Mono<Device> when canRegister returns empty mono for a given deviceId:

device.copy(status = DeviceStatus.REGISTRATION_FAILED)

Similarly I want to modify each emission of Flux<Device> for combination of:

deviceRepository.getDevices(userId) : Flux<Device>

and

registrationRepository.getAllRegistrationInfos(): Flux<DeviceRegistrationInfo>

Answer 1

Just using streams it is impossible to do what you want. Both streams and reactive streams are not design for this type of conditional logic. It is one of the biggest minuses to the framework. And it might be addressed when the pattern matching will come in java 13 or 14.

But we are not totally out of options. As Optional comes to a resque. My example is in Java but i'm sure it is enough for you to start.

It will also work for the Flux.

@Test
public void c(){
    //Mono<Optional<DeviceRegistrationInfo>> second = Mono.just(Optional.ofNullable(new DeviceRegistrationInfo()));
    Mono<Optional<DeviceRegistrationInfo>> second = Mono.just(Optional.empty());
    Mono<Device> device_ = Mono.just(new Device());

    second
            .map( info -> info.map( t -> device_) )  //you can put what you want in that map function and it will be executed only if optional is there
            .filter(Optional::isPresent)
            .flatMap(Optional::get)
            .subscribe(System.out::println);

}

Answer 2

Inspired by @piotr szybicki answer I've came up with following answer:

    val canRegister: Mono<Optional<PreRegistrationInfo>> = preRegistrationRepository.canRegister(deviceId)
    val foundDevice: Mono<Device> = deviceRepository.getDevice(deviceId, userId);

    return canRegister.flatMap {
        when {
            !it.isPresent -> foundDevice.map(this::overrideStatusIfNeeded)
            else -> foundDevice
        }
    }
}

private fun overrideStatusIfNeeded(it: Device) = if (it.status == DeviceStatus.ADDED) {
    it.copy(status = DeviceStatus.REGISTRATION_FAILED)
} else {
    it
}