Каков наилучший способ вызвать второе Mono без потери результата первого? - PullRequest
Новая
       12

Каков наилучший способ вызвать второе Mono без потери результата первого?

0 голосов
/ 14 ноября 2018

У меня есть код, который должен взаимодействовать с двумя реактивными зависимостями - сохранение в базе данных и публикация в потоке, каждый из которых возвращает Mono экземпляров. После этого мне нужно еще немного обработать - все с использованием исходного объекта, с которого я начал, что-то вроде этого: 

val myObject = Mono.just(“thing”);
myObject.flatMap(repository::save)
        .flatMap(stream::publish)
        .map(obj -> moreProcessing(obj));

Но проблема в том, что repository.save не возвращает Mono интересующего меня объекта, а также stream.publish.

Я могу достичь того, что хочу, с чем-то вроде: 

myObject.flatMap(obj -> repository.save(obj).then(Mono.just(obj))
        .flatMap(obj -> stream.publish(obj).then(Mono.just(obj))
        .map(obj -> moreProcessing(obj));

т.е. сопоставление с функциями зависимостей только для того, чтобы получить их поведение, а затем сопоставление прямо обратно, но это кажется странным - как будто я могу упустить какой-то лучший способ использовать API, который не дает подсказки, что я сопоставляю между типами, потому что это не то, что я использую flatMap, действительно.

Я также могу пойти с чем-то вроде: 

myObject.flatMap(obj ->
            repository.save(obj)
                .flatMap(x -> publisher.stream(obj))
                .map(x -> moreProcessing(obj)));

но это тоже не очень хороший способ, так как в даже относительно простых случаях это приведет к сложному к сопровождению коду.

Мысли

1 Ответ

0 голосов
/ 14 ноября 2018

У меня была похожая проблема, и я решил ее с помощью кода ниже.Я думаю, что важно не потеряться в цепочке методов любой ценой.Поскольку это очень легко переоценить создание конструкций, которые трудно читать. 

@Test
public void mergeMonos(){
    Mono<String> myObject = Mono.just("my object");
    Mono<String> savedToDB = saveToDb(myObject);
    Mono<String> savedToStream = saveToStream(myObject);


    //if you need to wait for the first two to complete
    Flux.concat(savedToDB, savedToStream, myObject)
            .last()
            .subscribe(System.out::println);

    //if you don't need to wait for the db and stream
    Flux.merge(savedToDB, savedToStream, myObject)
            .last()
            .subscribe(System.out::println);
}

private Mono<String> saveToStream(Mono<String> myObject) {
    return myObject.map(t-> "saved to stream");
}

private Mono<String> saveToDb(Mono<String> myObject) {
    return myObject.map(t-> "saved to db");
}
...