spring-boot - 如何连接嵌套通量

Question

如何将两个通量与嵌套通量连接起来？为什么这段代码执行永远不会结束？

@Test
fun `concatenating two flux`() {

    val names = listOf("israel", "israel")

    val a = Flux.just("a", "v")
            .flatMap { it.toUpperCase().toMono() }
            .concatWith { names.joinToString(" ").toMono() }

    StepVerifier.create(a).expectNext("A", "V", "israel israel").verifyComplete()
}

当我有带有分离变量的通量时，执行按预期运行

@Test
fun `concatenating two flux`() {

    val names = listOf("israel", "israel")

    val b = names.joinToString(" ").toMono()

    val a = Flux.just("a", "v")
            .flatMap { it.toUpperCase().toMono() }

    val c = a.concatWith(b)

    StepVerifier.create(c.log()).expectNext("A", "V", "israel israel").verifyComplete()
}

Answer 1

您需要使用()而不是{}在concatWith()

// RIGHT!

    Flux.just("a", "v")
                    .flatMap { it.toUpperCase().toMono() }
                    .concatWith ( names.joinToString(" ").toMono() )

// WRONG!

    Flux.just("a", "v")
                    .flatMap { it.toUpperCase().toMono() }
                    .concatWith { names.joinToString(" ").toMono() }

大多数 Rx2 方法确实接受实际的 lambda，有些方法接受 Callable<ObservableSource<T>>而不是ObservableSource ，其他人取Function<T, ObservableSource<R>> .

Observable.defer { Observable.just(1) } ——这会很好地工作。

或者 observable.flatMap { Observable.just(1) } - 也会按预期工作(如果你故意忽略传入参数)。

第三个事实是我们已经习惯了 Rx1，它总是需要 Observable在其 andThen()方法，它不能表示为 lambda ，这就是为什么我们需要使用 ()而不是{}