Tester le tampon et la fin d’AsyncStream sur un Mac cloud

Relier des callbacks de progression de téléchargement, d’état d’un appareil ou de notification à un AsyncStream ne suffit pas à garantir la fiabilité du flux, même si l’interface se met à jour normalement. Si le consommateur prend un peu de retard, le tampon peut perdre des valeurs ; lorsque l’utilisateur quitte une page, l’abonnement sous-jacent peut aussi rester actif. Sur un Mac cloud de VpsKeep, des tests Swift autonomes permettent de vérifier ces comportements avant de réintégrer le flux dans l’application, sans devoir multiplier les clics dans l’interface pour débusquer les conditions de concurrence.

Définir d’abord les événements qui peuvent être perdus

Pour un pourcentage de progression, seule la valeur la plus récente est généralement utile : bufferingNewest convient donc bien. Si les premiers événements reçus doivent être conservés, on peut choisir bufferingOldest. Aucune de ces stratégies ne constitue une file sans perte. Les données qui ne doivent pas être perdues, comme les modifications de commandes ou les confirmations d’écriture de fichiers, nécessitent un mécanisme supplémentaire d’acquittement ou de persistance.

Le tampon non borné utilisé par défaut n’est pas non plus une solution universelle. Si la production reste durablement plus rapide que la consommation, les éléments en attente occupent toujours plus de mémoire. Commencez par fixer trois contraintes : la capacité du tampon, les éléments à conserver lorsqu’il est plein et le moment où l’abonnement doit être libéré. La capacité doit correspondre au retard de traitement acceptable pour l’application, et non être augmentée arbitrairement en fonction de la mémoire encore disponible.

Avec AsyncStream, yield n’attend pas que le consommateur libère une place. Un tampon borné fournit une stratégie de rejet, pas une contre-pression automatique.

Cet article vérifie ces règles avec un flux d’entiers dont la capacité est de 2. Les entiers représentent les numéros des événements, pas des mesures de performance.

Créer un espace de travail de test autonome

Exécutez les commandes suivantes dans un répertoire accessible en écriture pour créer un paquet de test avec la chaîne d’outils de développement actuellement sélectionnée. Notez la version de Swift et conservez la même chaîne d’outils pour les exécutions ultérieures. Le paquet n’a besoin ni des clés secrètes de l’application ni d’une connexion à des services réels.

mkdir StreamContract
cd StreamContract
swift --version
swift package init --type library
mkdir -p Tests/StreamContractTests

Enregistrez le code de la section suivante dans Tests/StreamContractTests/StreamContractTests.swift. Si l’initialisation a créé un fichier du même nom, remplacez-le. Les autres tests générés par le modèle peuvent être conservés : le filtre utilisé plus bas ne sélectionne que cette classe.

Pour rendre le résultat du débordement déterministe, le test produit d’abord trois éléments de manière synchrone, puis démarre la consommation. Ne laissez pas le producteur et le consommateur se disputer l’ordonnancement, et n’utilisez pas une « attente d’une seconde » pour supposer que le tampon est plein. Ce qui est vérifié ici, c’est le contrat du conteneur, pas la vitesse d’exécution des threads.

Vérifier les valeurs rejetées et la séquence conservée

Une assertion sur le tableau final ne suffit pas. Le code doit également vérifier la valeur renvoyée par le troisième yield : lorsque les éléments les plus récents sont conservés, c’est l’ancien élément 1 qui est rejeté ; lorsque les premiers éléments sont conservés, c’est le nouvel élément 3 qui est rejeté. Après finish, les valeurs déjà présentes dans le tampon peuvent encore être lues jusqu’au bout.

import XCTest

final class StreamContractTests: XCTestCase {
    func testNewestKeepsLastTwo() async {
        let pair = AsyncStream<Int>.makeStream(
            bufferingPolicy: .bufferingNewest(2)
        )
        pair.continuation.yield(1)
        pair.continuation.yield(2)

        switch pair.continuation.yield(3) {
        case .dropped(let value):
            XCTAssertEqual(value, 1)
        default:
            XCTFail("Expected a dropped value")
        }

        pair.continuation.finish()
        var received: [Int] = []
        for await value in pair.stream {
            received.append(value)
        }
        XCTAssertEqual(received, [2, 3])

        switch pair.continuation.yield(4) {
        case .terminated:
            break
        default:
            XCTFail("Expected a terminated stream")
        }
    }

    func testOldestKeepsFirstTwo() async {
        let pair = AsyncStream<Int>.makeStream(
            bufferingPolicy: .bufferingOldest(2)
        )
        pair.continuation.yield(1)
        pair.continuation.yield(2)

        switch pair.continuation.yield(3) {
        case .dropped(let value):
            XCTAssertEqual(value, 3)
        default:
            XCTFail("Expected a dropped value")
        }

        pair.continuation.finish()
        var received: [Int] = []
        for await value in pair.stream {
            received.append(value)
        }
        XCTAssertEqual(received, [1, 2])
    }
}

Dans le code de production, le traitement de .dropped doit comptabiliser les rejets ou déclencher une compensation métier, plutôt qu’écrire un journal de manière synchrone pour chaque événement de progression. Le traitement de .terminated doit arrêter l’envoi des événements, au lieu de réessayer sans cesse avec la même continuation.

Vérifier que l’annulation déclenche le désabonnement

Ajoutez l’extension suivante dans le même fichier. Le flux ne contient aucun élément : le consommateur ne peut qu’attendre ou réagir à l’annulation. La notification de terminaison est vérifiée à l’aide d’une attente de test, sans temporisation fixe. Les deux secondes ne sont qu’un délai d’expiration en cas d’échec, pas une pause entre les étapes d’exécution.

extension StreamContractTests {
    func testCancellationReportsTermination() async {
        let ended = expectation(description: "Stream cancelled")
        let pair = AsyncStream<Int>.makeStream()

        pair.continuation.onTermination = { reason in
            switch reason {
            case .cancelled:
                ended.fulfill()
            case .finished:
                XCTFail("Unexpected normal finish")
            @unknown default:
                XCTFail("Unknown termination reason")
            }
        }

        let consumer = Task {
            for await _ in pair.stream {}
        }

        consumer.cancel()
        await consumer.value
        await fulfillment(of: [ended], timeout: 2)
    }
}

Dans l’adaptateur réel, l’appel qui valide l’attente de test doit être remplacé par la méthode de libération du jeton d’abonnement. Rien ne garantit que le traitement de la terminaison s’exécute sur le thread principal : il ne doit donc pas mettre directement à jour l’interface. Le désabonnement doit être sûr vis-à-vis des accès concurrents et idempotent. Évitez également d’appeler une fonction de désabonnement susceptible de déclencher à son tour un callback pendant qu’un verrou interne de l’adaptateur est détenu.

Il faut aussi prendre en compte une condition de concurrence lors de l’établissement de l’abonnement : la terminaison peut survenir avant que le jeton soit renvoyé. L’adaptateur doit mémoriser un état « fermé ». Si un jeton arrive ensuite et constate cet état, il doit être libéré immédiatement. Définir uniquement onTermination sans couvrir ce chemin laisserait subsister une fuite.

Intégrer les tests au pipeline et vérifier avant livraison

Exécutez les tests et conservez leur sortie dans les rapports de test habituels du pipeline. Avec des entrées fixes, chaque exécution doit produire les mêmes tableaux et rejeter les mêmes éléments. Un test qui ne réussit que de temps en temps indique généralement que des temporisateurs réels ou une source d’événements partagée se sont glissés dans le test.

swift test --filter StreamContractTests

Avant de réintégrer le flux dans l’application, vérifiez quatre points : un finish normal entraîne-t-il le désabonnement ? L’annulation de la tâche entraîne-t-elle le désabonnement ? Les callbacks tardifs sont-ils ignorés pendant le désabonnement ? Un jeton reçu après la terminaison est-il immédiatement libéré ? Testez d’abord ces chemins avec une fausse source d’événements permettant des envois manuels, puis avec la source réelle.

Ne simulez pas une diffusion à plusieurs destinataires en laissant plusieurs tâches se disputer la lecture du même flux. Si plusieurs abonnés sont nécessaires, concevez explicitement une couche de distribution et définissez séparément la stratégie de tampon de chacun. Conservez enfin le code des tests, la version de la chaîne d’outils et la description du contrat métier. Les personnes qui raccorderont ensuite une source d’événements sauront ainsi quelles pertes sont prévues par conception et lesquelles signalent un dysfonctionnement.

Questions fréquentes