Vert.X streams with Reactor

VertXAndReactor.kt

fun <Input, Output> processWithReactor(
  vertx: Vertx,
  inputStream: ReadStream<Input>,
  outputStream: WriteStream<Output>,
  processor: (Flux<Input>) -> Flux<Output>
) {
  val publisher = ReactiveWriteStream.writeStream<Input>(vertx)
  inputStream.pipeTo(publisher)

  val writeStreamFlux = Flux.from(publisher)
  val processedFlux = processor(writeStreamFlux)

  val subscriber = ReactiveReadStream.readStream<Output>()
  processedFlux.subscribe(subscriber)

  subscriber.pipeTo(outputStream)
}

// Use the helper in a Vert.X handler
processWithReactor(
  vertx,
  inputStream = websocketStream,
  outputStream = eventBusPublisher
) { socketFlux ->
  socketFlux
    .flatMap { /* do things*/ }
    .buffer(10)
}