Краткий ответ

Отмена потребляющей задачи приводит к вызову onTermination, когда AsyncStream обнаруживает отмену во время очередного ожидания элемента, обычно при выполнении следующего шага for await. Если задача в этот момент занята синхронной работой и не взаимодействует со stream, callback не обязан сработать немедленно.

onTermination также вызывается при явном завершении continuation или разрушении stream. Это уведомление о прекращении жизненного цикла потока, а не гарантия немедленной остановки producer-кода.

Исторический контекст

AsyncStream предназначен для адаптации push-моделей — callback API, делегатов и событий — к pull-модели AsyncSequence, где потребитель получает элементы через for await. Такой подход позволяет использовать структурированное ожидание и отмену вместо ручного управления множеством callback-цепочек.

При этом producer и consumer имеют разные жизненные циклы. onTermination появился как механизм, позволяющий producer узнать, что потребитель больше не будет получать элементы, и освободить связанные ресурсы.

Постановка проблемы

Представим подписку на системные события или длительный запрос к внешнему источнику. Пользователь закрывает экран, потребляющая задача отменяется, но producer продолжает получать события и удерживать ресурсы.

Ошибка состоит в предположении, что отмена задачи мгновенно прервет любой код и немедленно вызовет onTermination. Отмена в Swift кооперативная: stream должен наблюдать состояние отмены в подходящей точке, а producer должен корректно реагировать на сигнал завершения.

Подробное решение

onTermination принадлежит continuation и вызывается один раз при завершении stream. Основные причины — вызов finish, отмена итерации потребителем или прекращение существования внутреннего состояния stream после освобождения связанных объектов.

При отмене задачи, выполняющей for await, итератор обычно обнаруживает отмену на следующем ожидании элемента. Цикл завершается, а continuation получает уведомление через onTermination. Если задача выполняет долгую синхронную операцию и не дошла до следующего next(), автоматической немедленной реакции нет.

let stream = AsyncStream<Int> { continuation in
    continuation.onTermination = { reason in
        print("Источник остановлен: \(reason)")
        stopExternalSubscription()
    }

    startExternalSubscription { value in
        _ = continuation.yield(value)
    }
}

let consumer = Task {
    for await value in stream {
        process(value)
    }
}

consumer.cancel()

После отмены producer не должен продолжать бесконечно публиковать значения. Результат yield следует учитывать: после завершения stream новые элементы не будут доставлены, поэтому внешний источник можно остановить или удалить его регистрацию.

onTermination не следует использовать как единственный механизм синхронной остановки критически важной операции. Внешний API может иметь собственную задержку отмены, а обработчик может выполняться не на ожидаемом акторе или потоке. Если освобождение ресурса должно быть безопасным при повторных сигналах, оно должно быть идемпотентным и защищённым подходящим механизмом синхронизации.

Ситуация из практики

Экран подписывается на поток геолокационных событий через AsyncStream. При уходе со страницы задача-потребитель отменяется, но объект геолокации нельзя оставлять активным: он будет расходовать батарею и продолжит поставлять события.

Вариант с остановкой источника только после выхода из внешней функции ненадёжен: stream может жить дольше функции из-за захватов. Вариант с ручным вызовом finish при каждом возможном пути выхода лучше контролирует завершение, но легко пропустить путь отмены или вызвать завершение из нескольких мест.

Практичное решение — зарегистрировать onTermination, а остановку геолокации сделать идемпотентной. Отмена consumer-задачи приводит к завершению итерации, stream уведомляет producer, а producer прекращает подписку. Явное завершение через finish при нормальном закрытии источника остаётся полезным, потому что позволяет отличить штатное завершение от отмены.

Что кандидаты часто упускают

1. Обязательно ли onTermination вызывается в момент вызова Task.cancel()?

Нет. Task.cancel() только устанавливает состояние отмены задачи. AsyncStream обнаруживает его, когда итератор проверяет отмену, обычно во время ожидания следующего элемента. Поэтому при блокирующей синхронной работе consumer callback может быть отложен.

2. Означает ли вызов onTermination, что producer уже остановлен?

Нет. Это уведомление, а не принудительное уничтожение producer-кода. Обработчик должен самостоятельно отменить внешний запрос, удалить подписку или изменить состояние producer; если этого не сделать, источник может продолжать работу.

3. Можно ли безопасно считать onTermination заменой блокировке?

Нет. Callback сообщает о завершении жизненного цикла stream, но не делает произвольное общее состояние безопасным для конкурентного доступа. Если обработчик и producer одновременно изменяют ссылочный объект или реестр подписок, нужны actor, Mutex или другой корректный механизм синхронизации; также полезно сделать очистку идемпотентной.