Поток веб-клиента Java vertx с сервера с фрагментированными сообщениями ⇐ JAVA

[Anonymous]

У меня есть сервер vertx, который отправляет информацию о событиях всем стримерам.
Используя консоль, все работает так, как хотелось.
Но если я использую веб-клиент vertx, клиент подключается, но ничего не получает.
Веб-клиент

Код: Выделить всё

val webClientOpt = WebClientOptions()
.setKeepAlive(true)
.setUserAgent("Client/2.0")
.setFollowRedirects(true)
.setShared(true)

вызов клиента

Код: Выделить всё

client
.get(port, host, UriTemplate.of(s"${path}event-bus/"))
.putHeader("content-type", "application/json")
.bearerTokenAuthentication(UserBuffer.loggedInUser().jwtToken())
.as(BodyCodec.pipe(writeBuffer)).send()

теперь writeBuffer и Reader

Код: Выделить всё

val writeBuffer = ReactiveWriteStream.writeStream[Buffer](vertx)
val readStream = ReactiveReadStream.readStream[Buffer]()

Также для получения данных

Код: Выделить всё

readStream.handler(j => {
println("CONSUMING!!!!")
println(j.toString("UTF-8"))
})
writeBuffer.subscribe(readStream)

Я знаю, что мне нужно использовать BodyCodec.pipe, но думаю, вот в чем моя проблема. Думаю, я использую его неправильно.

Мне было непонятно, как правильно использовать WriteStream.
Теперь я использую

Код: Выделить всё

new WriteStream[Buffer]() {
override def write(buffer: Buffer): io.vertx.core.Future[Void] = {
println(buffer.toString())
Future.successful(null).asVertx
}
override def end(): io.vertx.core.Future[Void] = {
Future.successful(null).asVertx
}
override def setWriteQueueMaxSize(maxSize: Int): WriteStream[Buffer] = this
override def writeQueueFull(): Boolean = false
override def drainHandler(handler: Handler[Void]): WriteStream[Buffer] = this
override def exceptionHandler(handler: Handler[Throwable]): WriteStream[Buffer] = this
}

Это то, что я искал.


Подробнее здесь: https://stackoverflow.com/questions/797 ... d-messages