AsyncStream auf dem Cloud-Mac: Puffer und Abschluss testen

Wenn Downloadfortschritt, Gerätestatus oder Benachrichtigungs-Callbacks in einen AsyncStream eingespeist werden, bedeutet eine korrekt aktualisierte Oberfläche noch nicht, dass der Ereignisstrom korrekt arbeitet. Ist der Consumer etwas langsamer, kann der Puffer Werte verwerfen; beim Verlassen einer Seite kann das zugrunde liegende Abonnement weiterlaufen. Auf einem Cloud-Mac von VpsKeep lassen sich diese Verhaltensweisen zunächst mit eigenständigen Swift-Tests absichern, bevor der Stream wieder in die Anwendung eingebunden wird. So muss man Race Conditions nicht durch wiederholtes Klicken in der Oberfläche aufspüren.

Zuerst festlegen, welche Ereignisse verworfen werden dürfen

Bei Fortschrittsangaben in Prozent wird normalerweise nur der neueste Wert benötigt. Dafür eignet sich bufferingNewest. Müssen hingegen die zuerst eingegangenen Ereignisse erhalten bleiben, kommt bufferingOldest infrage. Beide Varianten sind keine verlustfreien Warteschlangen. Daten, die nicht verloren gehen dürfen, etwa Bestelländerungen oder Bestätigungen von Dateischreibvorgängen, benötigen zusätzliche Bestätigungs- oder Persistenzmechanismen.

Auch der standardmäßig unbegrenzte Puffer ist keine Universallösung. Werden dauerhaft mehr Elemente produziert als konsumiert, belegen die wartenden Elemente immer mehr Speicher. Halten Sie zunächst drei Vorgaben fest: die Pufferkapazität, welche Elemente bei vollem Puffer erhalten bleiben und wann das Abonnement freigegeben wird. Die Kapazität sollte sich nach dem fachlich akzeptablen Rückstau richten und nicht beliebig anhand des noch verfügbaren Arbeitsspeichers erhöht werden.

AsyncStream wartet bei yield nicht darauf, dass der Consumer Platz schafft. Ein begrenzter Puffer bietet eine Strategie zum Verwerfen von Elementen, keine automatische Backpressure.

In diesem Artikel werden die Regeln mit einem Integer-Stream der Kapazität 2 geprüft. Die Integer stehen für Ereignisnummern, nicht für Leistungsmesswerte.

Einen eigenständigen Test-Workspace anlegen

Führen Sie die folgenden Befehle in einem beschreibbaren Verzeichnis aus, um mit der aktuell ausgewählten Entwicklungstoolchain ein Testpaket zu erstellen. Dokumentieren Sie die Swift-Version und verwenden Sie bei späteren Wiederholungen dieselbe Toolchain. Das Paket benötigt weder Zugangsschlüssel der Anwendung noch eine Verbindung zu echten Diensten.

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

Speichern Sie den Code aus dem nächsten Abschnitt unter Tests/StreamContractTests/StreamContractTests.swift. Falls bei der Initialisierung bereits eine gleichnamige Datei angelegt wurde, ersetzen Sie sie. Andere von der Vorlage erzeugte Tests können bestehen bleiben; der unten verwendete Filter führt nur diese Klasse aus.

Damit das Ergebnis bei einem Pufferüberlauf eindeutig ist, erzeugt der Test zunächst synchron drei Elemente und beginnt erst danach mit dem Konsumieren. Lassen Sie Producer und Consumer nicht gleichzeitig um die Ausführung konkurrieren. Verwenden Sie auch kein „eine Sekunde warten“, um zu vermuten, dass der Puffer inzwischen voll ist. Geprüft wird hier der Vertrag des Containers, nicht die Ausführungsgeschwindigkeit von Threads.

Verworfene Werte und die erhaltene Reihenfolge prüfen

Nur das endgültige Array zu prüfen, reicht nicht aus. Der Code sollte auch den Rückgabewert des dritten yield prüfen: Werden die neuesten Elemente behalten, wird das alte Element 1 verworfen; werden die frühesten Elemente behalten, wird das neue Element 3 verworfen. Auch nach finish können die bereits gepufferten Werte noch vollständig gelesen werden.

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])
    }
}

Produktionscode sollte bei .dropped die Anzahl verworfener Elemente mitzählen oder eine fachliche Kompensation auslösen, statt für jedes Fortschrittsereignis synchron einen Logeintrag zu schreiben. Bei .terminated sollte er die weitere Zustellung einstellen, anstatt dieselbe Continuation immer wieder zu verwenden.

Prüfen, ob ein Abbruch die Abmeldung auslöst

Ergänzen Sie dieselbe Datei um die folgende Erweiterung. Der Stream enthält keine Elemente, sodass der Consumer nur warten oder auf einen Abbruch reagieren kann. Die Benachrichtigung über die Beendigung wird mit einer Testerwartung geprüft, nicht mit einer festen Wartezeit. Die zwei Sekunden sind lediglich das Timeout für einen fehlgeschlagenen Test, keine Pause zwischen Ausführungsschritten.

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)
    }
}

Im tatsächlichen Adapter sollte anstelle des Aufrufs zur Erfüllung der Testerwartung die Freigabemethode des Abonnement-Tokens aufgerufen werden. Es ist nicht garantiert, dass die Behandlung der Beendigung auf dem Hauptthread ausgeführt wird; die Oberfläche darf dort daher nicht direkt aktualisiert werden. Die Abmeldung muss threadsicher und idempotent sein. Rufen Sie außerdem keine Abmeldefunktion, die ihrerseits einen Callback auslösen könnte, auf, während eine interne Sperre des Adapters gehalten wird.

Berücksichtigen Sie auch eine Race Condition beim Aufbau des Abonnements: Die Beendigung kann eintreten, bevor das Token zurückgegeben wird. Der Adapter muss einen Zustand „geschlossen“ speichern. Trifft das Token später ein und ist der Zustand bereits geschlossen, muss es sofort freigegeben werden. Nur onTermination zu setzen, ohne diesen Pfad abzudecken, reicht nicht aus.

In die Pipeline einbinden und vor der Übergabe prüfen

Führen Sie die Tests aus und speichern Sie die Ausgabe in den regulären Testprotokollen der Pipeline. Bei festen Eingaben sollten wiederholte Durchläufe dieselben Arrays und verworfenen Elemente ergeben. Besteht ein Test nur gelegentlich, deutet das meist darauf hin, dass echte Timer oder gemeinsam genutzte Ereignisquellen in den Test geraten sind.

swift test --filter StreamContractTests

Prüfen Sie vor der erneuten Einbindung in die Anwendung vier Punkte: Wird bei normalem finish abgemeldet? Wird beim Abbruch des Tasks abgemeldet? Werden verspätete Callbacks während der Abmeldung ignoriert? Wird ein Token sofort freigegeben, wenn es erst nach der Beendigung eintrifft? Testen Sie diese Pfade zunächst mit einer künstlichen Ereignisquelle, in die sich Ereignisse manuell einspeisen lassen, und anschließend mit der echten Quelle.

Simulieren Sie keinen Broadcast, indem mehrere Tasks um das Lesen desselben Streams konkurrieren. Wenn mehrere Abonnenten benötigt werden, muss die Verteilungsschicht ausdrücklich dafür ausgelegt sein. Legen Sie außerdem für jeden Abonnenten eine eigene Pufferstrategie fest. Bewahren Sie abschließend den Testcode, die Toolchain-Version und die Beschreibung des fachlichen Vertrags auf. So wissen alle, die später eine Ereignisquelle anbinden, welche Verluste beabsichtigt sind und welche auf einen Fehler hinweisen.

Häufig gestellte Fragen