使用Spring 5 WebFlux框架解码字节数组

Question

我正在尝试使用新的Spring WebFlux框架和Kotlin。但是我无法找到这段代码（myService）出了什么问题：

fun foo(): Flux<ByteArray> {
    val client = WebClient.create("http://byte-array-service")
    return client
            .get()
            .uri("/info")
            .accept(MediaType.APPLICATION_OCTET_STREAM)
            .exchange()
            .flatMapMany {
                r -> r.bodyToFlux(ByteArray::class.java)
            }
}

这种方法返回的Flux大小为7893字节，我知道不是所有的字节都是由byte-array-service发送的。如果我使用旧式的rest template一切都正常。

fun foo(): Flux<ByteArray> {
    val rt = RestTemplate()
    rt.messageConverters.add(
            ByteArrayHttpMessageConverter())
    val headers = HttpHeaders()
    headers.accept = listOf(MediaType.APPLICATION_OCTET_STREAM)

    val entity = HttpEntity<String>(headers)
    val r = rt.exchange("http://byte-array-service/info", HttpMethod.GET,entity, ByteArray::class.java)
    return Flux.just(r.body)
}

它返回从byte-array-service发送的所有274124字节。

这是我的消费者。

fun doReadFromByteArrayService(req: ServerRequest): Mono<ServerResponse> {

    return Mono.from(myService
            .foo()
            .flatMap {
                accepted().body(fromObject(it.size))
            })
}

- Pavel Varchenko

---

那么在其他函数中，当您尝试从Flux<ByteArray>中使用数据时，您只能读取7893个字节，对吗？ - Orest

你可以尝试使用Spring的DataBuffer类型吗？从中提取字节数组很容易，并且您可以利用它的有趣属性（如零拷贝和缓冲池）。 - Brian Clozel

@Orest 是的，你是正确的。 - Pavel Varchenko

@PavelVarchenko，您能否也发布消费者函数？ - Orest

@Orest，是的。已更改问题描述。 - Pavel Varchenko

Answer 1

如果我理解您的问题正确，您只需要将流量向前传递，那么这应该可以解决问题。我在自己的环境中进行了测试，并且没有遇到读取所有字节的问题。

获取字节的方法：

fun foo(): Flux<ByteArray> =
    WebClient.create("http://byte-array-service")
        .get()
        .uri("/info")
        .accept(MediaType.APPLICATION_OCTET_STREAM)
        .retrieve()
        .bodyToFlux(ByteArray::class.java)

返回带有响应的字节：

fun doReadFromByteArrayService(req: ServerRequest): Mono<ServerResponse> =
        ServerResponse.ok().body(foo())  

- Alphaone