主页 / coding / 问题 / 77674891
Accepted
MartinHH
Asked: 2023-12-17 23:11:40 +0800 CST

fs2：流启动后如何执行某些操作（“doOnSubscribe”）？

  • 772

我试图在猫效应 IO-App 的上下文中使用不纯的（“java”）API。不纯的 API 看起来有点像这样：

import  io.reactivex.Flowable
import java.util.concurrent.CompletableFuture

trait ImpureProducer[A] {
  /** the produced output - must be subscribed to before calling startProducing() */
  def output: Flowable[A]

  /** Makes the producer start publishing its output to any subscribers of `output`. */
  def startProducing(): CompletableFuture[Unit]
}

（当然，还有更多方法，包括 stopProducing()，但这些与我的问题无关。）

我的（也许是幼稚的）适应该 API 的方法如下所示（利用Flowablea 的事实org.reactivestreams.Publisher）：

import cats.effect.IO
import fs2.Stream
import fs2.interop.reactivestreams.*

class AdaptedProducer[A](private val underlying: ImpureProducer[A]) {
  def output: Stream[IO, A] =
    underlying.output.toStreamBuffered(1)
  def startProducing: IO[Unit] = 
    IO.fromCompletableFuture(IO(underlying.startProducing()))
}

我的问题是：如何确保output在评估之前订阅 -stream startProducing？

例如，如何修复以下尝试以获取生成的第一个项目的 IO：

import cats.Parallel
import cats.effect.IO

def firstOutput[A](producer: AdaptedProducer[A]): IO[A] = {
  val firstOut: IO[A] = producer.output.take(1).compile.onlyOrError
  // this introduces a race condition: it is not ensured that the output-stream
  // will already be subscribed to when startProducing is evaluated.
  Parallel[IO].parProductL(firstOut)(producer.startProducing)
}
scala

1 个回答

  1. Best Answer

    这可以使用新的Java 互操作来完成flow。
    请注意，自从添加了反应流inteop 以来，它已被弃用flow。API 被重新设计以处理更多情况（如本例），并且从头开始重新实现实现以提高效率。
    如果您使用的 API 没有flow基础版本，您可以使用FlowAdapters来包装它。

    代码如下所示：

    import org.reactivestreams.FlowAdapters

final class AdaptedProducer[A](underlying: ImpureProducer[A], chunkSize: Int) {
  val run: Stream[IO, A] =
    fs2.interop.flow.fromPublisher(chunkSize) { subscriber =>
       IO(
         underlying.output.subscribe(
            FlowAdapters.toFlowSubscriber(
             subscriber
            )
         )
       ) >>
       IO.fromCompletableFuture(IO(underlying.startProducing()))
    }    
}

    这确保startProducing仅当您实际开始使用Stream并且在调用之后才调用subscribe。

    希望这有帮助：D

    • 2

相关问题