Как вернуть реактивный WebClient Kotlin Coroutines Flow in Spring

Question

Spring 5.2 принес поддержку сопрограмм Kotlin, Spring реактивный WebClient получил поддержку сопрограмм в расширениях Kotlin.

Я создал бэкэнд-сервис, который выставляет GET /posts как поток, проверьте коды здесь .

@GetMapping("")
fun findAll(): Flow<Post> =
        postRepository.findAll()

В примере клиента я пытался использовать WebClient для использования этого API следующим образом.

@GetMapping("")
suspend fun findAll(): Flow<Post> =
        client.get()
                .uri("/posts")
                .accept(MediaType.APPLICATION_JSON)
                .awaitExchange()
                .awaitBody()

Не удалось из-за сериализации Джексона типа Flow.

Из-за метода awaitXXX в вышеприведенном выражении я должен использовать модификатор suspend для этой забавы.

Но работает следующее, если я изменил тип тела на Любой, проверьте коды участников .

GetMapping("")
suspend fun findAll() =
        client.get()
                .uri("/posts")
                .accept(MediaType.APPLICATION_JSON)
                .awaitExchange()
                .awaitBody<Any>()

После прочтения сопрограмм Kotlin Spring ref doc, поток должен быть преобразован в поток сопрограмм Kotlin. Как бороться с типом возврата Flow и удалить suspend здесь?

Обновление : тип возврата изменен на Flow, проверьте последние исходные коды здесь, я думаю, что это может быть частью Spring 5.2.0.M2. Модификатор suspend необходим для двухэтапных операций с сопрограммами в webclient api, как объяснено ниже Себастьеном Делезом.

Answer 1

Первое, что нужно понять, это то, что возвращение Flow не требует использования функций приостановки для самого метода-обработчика. При Flow функции приостановки обычно изолированы в параметрах лямбды. Но в этом (общем) случае использования, из-за API-интерфейса WebClient 2 (сначала получите ответ, затем получите тело), ​​нам нужно, чтобы метод обработчика был приостановлен для awaitExchange, а затем получил тело как Flow с bodyToFlow расширение:

@GetMapping("")
suspend fun findAll() =
    client.get()
        .uri("/posts")
        .accept(MediaType.APPLICATION_JSON)
        .awaitExchange()
        .bodyToFlow<Post>()

Поддерживается в Spring Framework 5.2 M2 и Spring Boot 2.2 M3 (см. связанную проблему ). Смотрите также мой связанный подробный пост в блоге .