诚然，这只是心理上的编译，但像这样的东西应该工作：

sourceOfAddresses
  .statefulMap(() => Option.empty[KillSwitch])(
    { (lastEmittedKillswitch, address) =>
      lastEmittedKillswitch.foreach(_.shutdown)

      // this just happens to match the desired ordering of the tuple, woohoo!
      buildSourceForAddress(address)
        .viaMat(Killswitches.single)(Keep.right)
        .preMaterialize()  // will need the materializer in implicit scope for this to work
    },
    _ => None
  )
  .detach // in combination with the shutdown above, ensure perpetual demand
  .flatMapConcat(identity)

基本上：

• 用一个XNUMX开关构建您传递给flatMapConcat的每个源，并预物质化该源，以便您可以访问XNUMX开关
• 在传递到flatMapConcat之前，保存该交换机
• 当你得到一个新的地址，如果有一个保存的交换机，激活它

触发开关将完成发出的源，导致flatMapConcat请求另一个源，该源在detach（基本上是一个缓冲区1）中随时可用（因为它是在触发触发开关后立即发出的）。由于detach现在是空的，因此通过statefulMap向sourceOfAddresses发送需求信号。