kotlin - Spring WebFlux Webclient 作为 Mono 接收应用程序/八位字节流文件

Question

我正在 Kotlin 中制作一个小型 Spring WebFlux 应用程序的原型(prototype)。该应用程序需要从远程 REST 端点获取 tar 存档并将其存储在本地磁盘上。听起来很简单。

我首先创建了一个集成测试，该测试启动 spring 服务器和另一个 WebFlux 服务器，并使用提供 tar 存档服务的模拟 REST 端点。

测试应该是这样的:

1) 应用程序:GET 模拟服务器/存档

2) 模拟服务器:响应状态为 200，正文中的 tar 存档作为类型附件

3) 应用程序:阻塞直到收到所有字节，然后解压并使用文件

我遇到的问题是，当我尝试将字节收集到应用程序上的 ByteArray 中时，它会永远阻塞。

我的mock-server/archive路由到以下函数:

fun serveArchive(request: ServerRequest): Mono<ServerResponse> {
    val tarFile = FileSystemResource(ARCHIVE_PATH)
    assert(tarFile.exists() && tarFile.isFile && tarFile.contentLength() != 0L)
    return ServerResponse
            .ok()
            .contentType(MediaType.APPLICATION_OCTET_STREAM)
            .contentLength(tarFile.contentLength())
            .header("Content-Disposition", "attachment; filename=\"$ARCHIVE_FNAME\"")
            .body(fromResource(tarFile))
}

然后我的应用程序使用以下内容调用它:

private fun retrieveArchive {
    client.get().uri(ARCHIVE_URL).accept(MediaType.APPLICATION_OCTET_STREAM)
            .exchange()
            .flatMap { response ->
                storeArchive(response.bodyToMono())
            }.subscribe()
}

private fun storeArchive(archive: Mono<ByteArrayResource>): Mono<Void> {
    val archiveContentBytes = archive.block() // <- this blocks forever
    val archiveContents = TarArchiveInputStream(archiveContentBytes.inputStream)
    // read archive
}

我看到了How to best get a byte array from a ClientResponse from Spring WebClient?这就是我尝试使用 ByteArrayResource 的原因。

当我逐步完成所有内容时，我发现 serveArchive 似乎正在工作(断言语句表示我传递的文件存在并且其中有一些字节)。在 retrieveArchive 中，我得到了 200，并且可以在 .headers 中看到所有适当的信息(内容类型、内容长度看起来都不错)。当我开始使用 storeArchive 并尝试使用 block 从 Mono 检索字节时，它只是永远阻塞。

我完全不知道如何调试这样的东西。

Answer 1

您只需从 flatMap 返回转换后的正文即可所以它从 Mono<T> 转变而来至T :

client.get().uri(ARCHIVE_URL).accept(MediaType.APPLICATION_OCTET_STREAM)
            .exchange()
            .flatMap { response ->
                response.bodyToMono(ByteArrayResource::class.java)
            }
            .map { archiveContentBytes ->
                archiveContentBytes.inputStream
            }
            .doOnSuccess { inputStream ->
                //here is you code to do anything with the inputStream
                val archiveContents = TarArchiveInputStream(inputStream)
            }
            .subscribe()