Creating a Monix Observable from a NuProcess ByteBuffer

Question

So I have a current project that uses Java Process and I am trying to replace it with NuProcess (i.e. https://github.com/brettwooldridge/NuProcess). For dealing with Java Process's STDOUT/STDERROR you have had an InputStream and since Monix provided a handy interopt method Observable.fromInputStream, this allowed you to easily create an Observable[String]/Observable[Array[Byte]].

However the issue is that NuProcess doesn't work with InputStream, instead it uses javas NIO ByteBuffer. Ontop of this its slightly more complex because it uses an event handling mechanism, i.e. to listen to STDOUT/STDERROR in Monix task for NuProcess you have to do something like

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
    }
  }
})

So the question is how would you hook this up to an observable (i.e. an Observable[String] or Observable[Array[Byte]] without being inefficient?

Note that I am using Monix 3.x

Creating a Monix Observable from a NuProcess ByteBuffer

0 Answers0