Котлин ждет Channel.isClosedForReceive

После звонка Channel.close() как лучше всего дождаться, пока Channel.isClosedForReceive станет правдой?

Я обрабатываю сообщения по порядку и хочу вернуть максимальное количество сообщений, обработанных после вызова Channel.close(). Однако, если я просто возьму максимальное обработанное сообщение после вызова close(), могут быть некоторые сообщения в канале до того, как будет использован «близкий токен», в результате чего реальное максимальное обработанное сообщение будет больше, чем возвращаемое значение.

Основываясь на документе для Channel.close(), я думаю, что Channel.isClosedForReceive - это то, чего мне следует ждать. Но я ожидал, что какая-то функция приостановки подождет, а не будет опрашивать ее статус.

/**
 * Immediately after invocation of this function
 * [isClosedForSend] starts returning `true`. However, [isClosedForReceive][ReceiveChannel.isClosedForReceive]
 * on the side of [ReceiveChannel] starts returning `true` only after all previously sent elements
 * are received.
 **/

kotlin kotlinx.coroutines
person Jake Walsh    schedule 27.02.2018    source источник

Ответы (1)


arrow_upward
3
arrow_downward

Предположим, вы обрабатываете все свои сообщения каким-то актером:

val channel = actor<Message> { ... }

Лучший способ узнать последнее сообщение, обработанное этим субъектом, - это создать явный обратный канал для передачи этой информации. Поскольку нам нужно передать только одно значение только один раз, мы будем использовать CompletableDeferred для этой цели:

val lastProcessed = CompletableDeferred<Message?>()

Теперь вы можете написать свой актор обработки сообщений следующим образом:

val actor = actor<Message> {
    var last: Message? = null
    try {
        for (msg in channel) {
            // process message
            last = msg
        }
    } finally {
        // report the last processed message via back channel
        lastProcessed.complete(last)
    }
}

Вот код, который закрывает канал и узнает последнее обработанное сообщение:

actor.close()
val last = lastProcessed.await() // receive last processed message
person Roman Elizarov    schedule 27.02.2018