第一个元素有时不包括在通量的第二个参数中。先切换？

Question

我试着使用电抗器中的SwitchOnFirst算子，这很神奇-除了有时作为双函数的第二个参数传递的transformer似乎不包括第一个元素。基本上，客户端通过RSocket向服务器发送2个项目。代码服务器端如下所示：

val socket = new AbstractRSocket() {

    override def requestChannel(payloads: Publisher[Payload]): Flux[Payload] =
      Flux.from(payloads).log.switchOnFirst((signal, all) => handle(signal.get(), all))

    private def handle(first: Payload, all: Flux[Payload]): Flux[Payload] =
      extractRoute(first) match {
        case Some("test.route") =>
          val source = Source.fromPublisher(all.log()).map(_.getDataUtf8)
          actorSink.runWith(source)
          return Flux.from(actorSource).map(DefaultPayload.create).runWith(Sink.asPublisher(false)))
      }

  }

客户端第一次启动时，服务器接收这两个项目，并按预期将其发布到actorsink。

[2020-02-01 16:17:42,656] [INFO] [reactor.Flux.DoFinallyFuseable.1] [] [reactor-tcp-epoll-2] - | onSubscribe([Fuseable] FluxDoFinally.DoFinallyFuseableSubscriber) {}
[2020-02-01 16:17:42,658] [INFO] [reactor.Flux.DoFinallyFuseable.1] [] [reactor-tcp-epoll-2] - | request(1) {}
[2020-02-01 16:17:42,664] [INFO] [reactor.Flux.DoFinallyFuseable.1] [] [reactor-tcp-epoll-2] - | onNext(io.rsocket.util.ByteBufPayload@53e655e6) {}
[2020-02-01 16:17:42,731] [INFO] [reactor.Flux.SwitchOnFirstInner.2] [] [akka.actor.default-dispatcher-9] - onSubscribe(FluxSwitchOnFirst.SwitchOnFirstInner) {}
[2020-02-01 16:17:42,736] [INFO] [reactor.Flux.SwitchOnFirstInner.2] [] [akka.actor.default-dispatcher-9] - request(16) {}
[2020-02-01 16:17:42,739] [INFO] [reactor.Flux.SwitchOnFirstInner.2] [] [akka.actor.default-dispatcher-9] - onNext(io.rsocket.util.ByteBufPayload@53e655e6) {}
[2020-02-01 16:17:42,741] [INFO] [reactor.Flux.DoFinallyFuseable.1] [] [akka.actor.default-dispatcher-9] - | request(15) {}
[Sink] Received (item1)
[2020-02-01 16:17:42,769] [INFO] [reactor.Flux.DoFinallyFuseable.1] [] [reactor-tcp-epoll-2] - | onNext(io.rsocket.util.ByteBufPayload@50215db3) {}
[2020-02-01 16:17:42,769] [INFO] [reactor.Flux.SwitchOnFirstInner.2] [] [reactor-tcp-epoll-2] - onNext(io.rsocket.util.ByteBufPayload@50215db3) {}
[Sink] Received (item2)
[2020-02-01 16:17:42,770] [INFO] [reactor.Flux.DoFinallyFuseable.1] [] [reactor-tcp-epoll-2] - | onComplete() {}
[2020-02-01 16:17:42,771] [INFO] [reactor.Flux.SwitchOnFirstInner.2] [] [reactor-tcp-epoll-2] - onComplete() {}
[Sink] Completed

但是，如果我停止客户端并再次运行它，则只发布第二项。

[2020-02-01 16:18:13,746] [INFO] [reactor.Flux.DoFinallyFuseable.3] [] [reactor-tcp-epoll-3] - | onSubscribe([Fuseable] FluxDoFinally.DoFinallyFuseableSubscriber) {}
[2020-02-01 16:18:13,746] [INFO] [reactor.Flux.DoFinallyFuseable.3] [] [reactor-tcp-epoll-3] - | request(1) {}
[2020-02-01 16:18:13,747] [INFO] [reactor.Flux.DoFinallyFuseable.3] [] [reactor-tcp-epoll-3] - | onNext(io.rsocket.util.ByteBufPayload@5a2d7823) {}
[2020-02-01 16:18:13,751] [INFO] [reactor.Flux.SwitchOnFirstInner.4] [] [akka.actor.default-dispatcher-6] - onSubscribe(FluxSwitchOnFirst.SwitchOnFirstInner) {}
[2020-02-01 16:18:13,752] [INFO] [reactor.Flux.SwitchOnFirstInner.4] [] [akka.actor.default-dispatcher-6] - request(16) {}
[2020-02-01 16:18:13,752] [INFO] [reactor.Flux.DoFinallyFuseable.3] [] [akka.actor.default-dispatcher-6] - | request(16) {}
[2020-02-01 16:18:13,787] [INFO] [reactor.Flux.DoFinallyFuseable.3] [] [reactor-tcp-epoll-3] - | onNext(io.rsocket.util.ByteBufPayload@1fa7bb46) {}
[2020-02-01 16:18:13,788] [INFO] [reactor.Flux.SwitchOnFirstInner.4] [] [reactor-tcp-epoll-3] - onNext(io.rsocket.util.ByteBufPayload@1fa7bb46) {}
[Sink] Received (item2)
[2020-02-01 16:18:13,790] [INFO] [reactor.Flux.DoFinallyFuseable.3] [] [reactor-tcp-epoll-3] - | onComplete() {}

一个区别是[SwitchOnFirstInner.4]请求（16）触发[DoFinallyFuseable.3]请求（16），而不是onNext使用SwitchOnFirst运算符中已经可用的第一个项目。

我可能做错了什么，但不知道是什么。SwitchOnFirst的javadoc指出，在所有情况下都应返回源自原始Flux的发布者，这里不是这种情况（输入发送到ActorSink，输出来自单独的ActorSource），可能是问题吗？

我是新来的Reactor/Reactor插座，所以如果我错过了一些明显的东西，很抱歉。

Answer 1

没有从allFlux派生返回Flux可能确实是原因，特别是如果actorSink.run（source）请求allsource，因为SwitchOnFirst运算符应该负责处理请求。