如何在通量流中获取实际值，而不是 FluxLift？

Question

我得到一个 Publisher<DataBuffer> inputStream 作为参数。我想将该流转换为一个字符串，记录该字符串，然后我必须将该字符串作为 Publisher 再次传递给另一个方法。

示例：

public Flux<Object> decode(Publisher<DataBuffer> inputStream) {
    return DataBufferUtils.join(inputStream)
                .map(buffer -> StandardCharsets.UTF_8.decode(buffer.asByteBuffer()).toString())
                .doOnNext(arg -> LOGGER.info(arg))
                .map(arg -> library.delegate(Mono.fromSupplier(() -> arg))) 
                .flatMapIterable(arg -> {
                     System.out.println(arg); //instanceof FluxLift??
                     return List.of(arg);
                );
}

class ExternalLibrary {
    //this ALWAYS returns a FluxLift
    public Flux<Object> delegate(Publisher<String> inputString) {
        //does not matter, I don't have control of this.
        //lets assume it does the following:

        return Flux.just(input).
            flatMapIterable(buffer -> List.of("some", "result"))
            .map(arg -> arg);
    }
}

问题：为什么最后的flatMapInterable()中的参数总是FluxLift类型？而且：return这里怎么会有真正的价值？

Answer 1

why is the argument in the final flatMapInterable() always of type FluxLift?

因为你的地图函数returns Flux

.map(arg -> library.delegate(Mono.fromSupplier(() -> arg)))

how can return the real value here?

映射函数 returns 反应类型时，使用 flatMap* 函数之一而不是 map。 flatMapMany 适合您的情况：

public Flux<Object> decode(Publisher<DataBuffer> inputStream) {
    return DataBufferUtils.join(inputStream) //Mono<DataBuffer>
                .map(buffer -> StandardCharsets.UTF_8.decode(buffer.asByteBuffer()).toString()) //Mono<String>
                .doOnNext(arg -> LOGGER.info(arg)) //Mono<String>
                .flatMapMany(arg -> library.delegate(Mono.fromSupplier(() -> arg))) // Flux<Object>
                .flatMapIterable(arg -> {
                     System.out.println(arg); // instanceof Object
                     return List.of(arg);
                );

如何在通量流中获取实际值，而不是 FluxLift？

How to get teh real value inside a flux flow, instead of FluxLift?

spring

reactor

spring-webflux