A modern programming language that makes developers happier.

kotlinlang

I have a coroutines-based API (ktor) that uses `ByteReadChannel` that I need to adapt to a rx2 API that uses `Flowable&lt;ByteBuffer&gt;`. Here are the methods I've come up with, with some guidance from <@U0BLPSJAK> and <@U096EGZ6U> on <#C0A974TJ9|ktor>, to adapt these in both directions -- these seem to work but opinions on correctness, efficiency, and style are welcome:

Adapting `Flowable&lt;ByteBuffer&gt;` to `ByteReadChannel`

```
suspend fun Flowable&lt;ByteBuffer&gt;.toByteReadChannel(parent: Job = Job()) = writer(Unconfined, parent = parent, autoFlush = true) {
  try {
    this@toByteReadChannel.consumeEach {
      channel.writeFully(it)
    }
  } catch (e: Throwable) {
    channel.close(e)
  }
}.channel
```

and in the other direction:

```
suspend fun ByteReadChannel.toFlowable(): Flowable&lt;ByteBuffer&gt; {
  return rxFlowable {
    this@toFlowable.lookAheadSuspend {
      consumeEachRemaining {
        // we have to make a copy as the ByteBuffer here comes from a reused ring buffer
        this@rxFlowable.channel.send(it.copy())
        true
      }
      this@rxFlowable.close()
    }
  }
}
```

Note the `copy()` in the second code snippet. I don't like this but don't know how to get rid of it.