Что эквивалентно методу трубы rxjs в пружинном реакторе?

Question

В Rx 5.5 есть метод канала, который позволяет применять пользовательский оператор к потоку.Есть ли подобный эквивалент для пружинного реактора?

Answer 1

Это не прямой эквивалент, но вы можете написать повторно используемые части цепочек операторов, написав Function<Flux, Publisher>, который вы передаете либо compose (применяя функцию для каждого входящего абонента, учитывая динамическое поведение вfunction) или transform (применяя функцию во время создания цепочки операторов Flux, один раз).

Например:

Function<Flux<String>, Flux<Integer>> letterPerWordCount = f -> f
    .flatMap(sentence -> Flux.fromArray(sentence.split(" "))
    .map(String::length);

Flux<String> mySentenceSource;
Flux<Integer> letterCount = mySentenceCount.transform(letterPerWordCount);

Answer 2

Похоже, что эти две функции могут быть использованы для этой цели:

public final <V> Flux<V> compose(Function<? super Flux<T>, ? extends Publisher<V>> transformer) 


public final <V> Mono<V> compose(Function<? super Mono<T>, ? extends Publisher<V>> transformer)