An example of using AsyncStream

PollingDataModel.swift

import Combine
import Foundation
import os

@MainActor public final class PollingDataModel<DataModels: Sendable> {
    private let logger = Logger(subsystem: "Polling", category: "PollingDataModel")

    private var pollingTask: Task<Void, Never>?
    private var currentlyPolling = false

    public init() {}

    public func startPolling(
        interval: TimeInterval = 10,
        triggerInitially: Bool = false,
        dataModels: [DataModels],
        action: @escaping ([DataModels]) async -> ([DataModels], TimeInterval?),
        update: @escaping ([DataModels]) -> Void
    ) {
        guard !currentlyPolling else { return }
        currentlyPolling = true

        pollingTask = Task {
            for await dataModels in await polling(
                interval: interval,
                triggerInitially: triggerInitially,
                dataModels: dataModels,
                action: action
            ) {
                update(dataModels)
            }
        }
    }

    public func stopPolling() {
        guard currentlyPolling else { return }
        pollingTask?.cancel()
        currentlyPolling = false
    }

    private func polling(
        interval: TimeInterval?,
        triggerInitially: Bool = false,
        dataModels: [DataModels],
        action: @escaping ([DataModels]) async -> ([DataModels], TimeInterval?)
    ) async -> AsyncStream<[DataModels]> {
        AsyncStream { continuation in
            var dataModels = dataModels
            var interval = interval

            Task {
                if triggerInitially {
                    (dataModels, interval) = await action(dataModels)
                }

                if let currentInterval = interval {
                    let intervals = Timer.publish(every: currentInterval, on: .main, in: .common)
                        .autoconnect()
                        .values

                    for await _ in intervals {
                        (dataModels, interval) = await action(dataModels)
                        continuation.yield(dataModels)
                    }
                }
            }

            return continuation.finish()
        }
    }
}