Изменить Mono, если другое Mono пусто в проекте Reactor?

Question

У меня есть Mono, который я хочу изменить, когда другой Mono пуст.

Первый Mono:

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

Второй Mono:

registrationRepository.canRegister(deviceId) : Mono<DeviceRegistrationInfo>

Чего я хочу добиться, так это изменить значение, возвращаемое Mono<Device>, когда canRegister возвращает пустое моно для данного deviceId:

device.copy(status = DeviceStatus.REGISTRATION_FAILED)

Аналогичным образом я хочу изменить каждый выбросFlux<Device> для комбинации:

deviceRepository.getDevices(userId) : Flux<Device>

и

registrationRepository.getAllRegistrationInfos(): Flux<DeviceRegistrationInfo>

Answer 1

Вдохновленный ответом @piotr szybicki Я пришел со следующим ответом:

    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
}

Answer 2

Просто используя потоки, невозможно делать то, что вы хотите.И потоки, и реактивные потоки не предназначены для этого типа условной логики.Это один из самых больших минусов фреймворка.И это может быть решено, когда сопоставление с образцом придет в java 13 или 14.

Но у нас не совсем нет вариантов.По желанию приходит в реск.Мой пример на Java, но я уверен, что вам достаточно начать.

Это также будет работать для 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);

}