scala - 从 NuProcess ByteBuffer 创建 Monix Observable

Question

所以我有一个使用 Java 的当前项目，Process我试图用 NuProcess 替换它（即https://github.com/brettwooldridge/NuProcess）。为了处理 JavaProcess的 STDOUT/STDERROR，您有一个InputStream并且由于 Monix 提供了一个方便的 interopt 方法Observable.fromInputStream，这使您可以轻松地创建一个Observable[String]/ Observable[Array[Byte]]。

但是问题是它NuProcess不适用于InputStream，而是使用 javas NIO ByteBuffer。除此之外，它稍微复杂一些，因为它使用事件处理机制，即在 Monix 任务中收听STDOUT/因为你必须执行类似的操作STDERRORNuProcess

nuProcess.setProcessHandler(new NuProcessHandler {
  override def onStderr(buffer: ByteBuffer, closed: Boolean) = {
    if (!closed) {
      val bytes = new Array[Byte](buffer.remaining())
      buffer.get(bytes)
      // Do something else here
    }
  }

  override def onStdout(buffer: ByteBuffer, closed: Boolean) = {
    if (!closed) {
      val bytes = new Array[Byte](buffer.remaining())
      buffer.get(bytes)
      // Do something else here
    }
  }
})

所以问题是你如何将它连接到一个可观察的（即一个Observable[String]或Observable[Array[Byte]]没有效率低下？

请注意，我使用的是 Monix 3.x