Блог

Я работаю над реактивными потоками, используя Spring webflux. Я хочу извлечь переменную( name ) из середины реактивного конвейера и использовать ее в другом месте следующим образом.

 public class Example {

    public Mono<String> test() {
        String name;

        return Mono.just("some random string")
                    .map(s -> {
                        name = s.toUpperCase(); 
                        return name;
                    }).map(...)
                      .map(...)
                      .flatMap(...)
                      .map(...)
                      .map(result -> result name)
                      .doOnSuccess(res -> asyncPublish(name));
     
     public void asyncPublish(String name) {
        // logic to write to a Messaging queue asynchronously
     }
    }
}
 

Приведенный выше код не работает. Это надуманный пример, но он показывает, чего я хочу достичь.

Примечание: Я не хочу использовать несколько zip операторов только для того, чтобы полностью перенести name карту на последнюю карту, где я хочу ее использовать. Есть ли способ сохранить его в переменной, как показано выше, а затем использовать его где-нибудь еще, где бы он мне ни понадобился.

Вы можете использовать, например, Tuple2 для передачи значения name вместе с измененными данными по цепочке.

 return Mono.just("some random string")
            .map(s -> s.toUpperCase())
            .map(s -> Tuples.of(s, x(s))) // given that x(s) is the transformation of this map statement
            .map(...) // keeping Tuple with the value of `name` in the first slot...
            .flatMap(...) // keeping Tuple with the value of `name` in the first slot...
            .map(resultTuple -> Tuples.of(resultTuple.getT1(), resultTuple.getT2()   resultTuple.getT1()) // keeping Tuple with the value of `name` in the first slot...
            .doOnSuccess(resultTuple -> asyncPublish(resultTuple.getT1()))
            .map(resultTuple -> resultTuple.getT2()); // in case that the returned Mono should contain the modified value...
 

Tuples находится в упаковке reactor.util.function и является частью активной зоны реактора.

Другим способом (без передачи значения по цепочке с использованием кортежей) может быть использование AtomicReference (но я все еще думаю, что способ кортежа чище). Способ атомарной привязки может выглядеть следующим образом:

 public Mono<String> test() {
    final AtomicReference<String> nameRef = new AtomicReference<>();

    return Mono.just("some random string")
                .map(s -> {
                    final String name = s.toUpperCase(); 
                    nameRef.set(name);
                    return name;
                }).map(...)
                  .map(...)
                  .flatMap(...)
                  .map(...)
                  .map(result -> result nameRef.get())
                  .doOnSuccess(res -> asyncPublish(nameRef.get()));
 
 public void asyncPublish(String name) {
    // logic to write to a Messaging queue asynchronously
 }
}