Spring WebFlux Webclient接收作为单声道的应用程序/八位组流文件

Question

我正在Kotlin中构建一个小型Spring WebFlux应用程序的原型。这个应用程序需要从远程RESTendpoint获取一个tar归档文件，并将其存储在本地磁盘上。听起来很简单。

我首先创建了一个集成测试，它启动spring服务器和一个带有模拟RESTendpoint的WebFlux服务器，该endpoint为tar存档提供服务。

测试应该是这样的：

1）app:GETmock-server/archive

2）mock-server：状态为200的响应，正文中的tar存档类型为attachment

我遇到的问题是，当我尝试将字节收集到应用程序上的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
}

我已经看到了如何最好地从Spring WebClient的ClientResponse中获取字节数组？这就是我试图使用ByteArrayResource的原因。

当我遍历所有内容时，我看到servearchive似乎正在工作（assert语句表示我正在传递的文件存在，并且其中有一些字节）。在retrievearchive中，我得到一个200，并且可以在.headers中看到所有适当的信息（内容类型、内容长度看起来都很好）。当我进入storearchive并尝试使用block从单声道中检索字节时，它只是永远阻塞。

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

Answer 1

您只需从flatmap返回转换后的主体，以便它从mono 转换为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()