FlowKit

Kotlin Flow semantics in Swift, with Swift-idiomatic API.

Full API documentation: https://developerbeau.github.io/FlowKit/documentation/flow/

FlowKit brings the Kotlin Flow mental model to Swift: cold asynchronous streams (Flow, ThrowingFlow), hot multicast primitives (StateFlow, SharedFlow), a rich operator library, and built-in testing infrastructure with virtual time.

FlowKit borrows Kotlin Flow's design, but it exists as a Swift library because of what the compiler checks before any of this code runs.

Every target in this package builds under Swift 6 language mode with strict concurrency enabled. The compiler tracks whether a value crossing a thread or an actor boundary is safe to share, and rejects the build when it isn't. Mutex is the clearest example: Value must conform to Sendable to instantiate the lock at all, and withLock's closure must return a Sendable result, so neither what goes into the lock nor what comes back out of it can be a value that was only safe to touch from one place. Actor isolation works the same way: touching actor-isolated state from the wrong context is a compile error, not a race that shows up under load.

Kotlin's coroutines don't check any of this at compile time. A Flow collector or a StateFlow update can cross threads with nothing in the type system confirming the value involved is safe to share, so a data race there is a runtime problem, caught by whichever test happens to exercise the right interleaving, or not caught at all. That's the tradeoff Swift's strict concurrency settings make: a class of concurrency bug becomes a compiler error where the code was written, instead of a crash report from wherever it happened to run.

Note on older Xcode: FlowKit uses Swift 6.2 and 6.3 language features (@concurrent, @specialize, isolated deinit, Observations async sequence). If you're on Xcode 17–26.3, install the Swift 6.3 toolchain from swift.org/install and select it in Xcode's Toolchains menu.

FlowKit's non-UI targets — the Flow and FlowTesting products — cross-compile for Android with the Swift SDK for Android:

swift build --swift-sdk aarch64-unknown-linux-android28 --target Flow --target FlowTesting

Two different things are checked, and they are not the same guarantee:

• Every pull request cross-compiles both products for aarch64 and x86_64 at API level 28. This is a required check: a change that breaks the Android build cannot merge. It is a build gate — nothing is executed.
• Nightly, and not required to pass, the eight non-UI test suites run on an x86_64 Android emulator. This is where FlowKit's runtime behaviour on Android is actually exercised. Because the suite is wall-clock-timing-sensitive and an emulator on a shared runner is not, a red night does not block anything and is not by itself a bug report.

So: the Android build is guaranteed; Android runtime behaviour is monitored, not gated. If you are shipping FlowKit on Android, treat the nightly emulator run as supporting evidence rather than a support commitment, and test on your own devices.

The FlowUI product is Apple-only. It compiles on Android — canImport(SwiftUI) and canImport(UIKit) are simply false there — so a green build of it would mean empty modules, not a working UI layer.

Add FlowKit to your Package.swift dependencies:

dependencies: [
    .package(url: "https://github.com/DeveloperBeau/FlowKit", from: "2.0.0")
]

Then add Flow to production targets and FlowTesting to test targets:

.target(
    name: "MyApp",
    dependencies: [
        .product(name: "Flow", package: "FlowKit")
    ]
),
.testTarget(
    name: "MyAppTests",
    dependencies: [
        "MyApp",
        .product(name: "FlowTesting", package: "FlowKit")
    ]
)

let articles: ThrowingFlow<[Article]> = ThrowingFlow { collector in
    // Emit cached articles first so the UI has something to show
    let cached = try await articleDatabase.fetchAll()
    try await collector.emit(cached)

    // Then fetch fresh articles from the network
    let fresh = try await articleAPI.fetchLatest()
    try await articleDatabase.upsert(fresh)
    try await collector.emit(fresh)
}

try await articles
    .map { articles in articles.filter { !$0.isArchived } }
    .collect { visibleArticles in
        articleListView.display(visibleArticles)
    }

enum SessionState: Sendable, Equatable {
    case signedOut
    case signingIn
    case signedIn(User)
    case error(String)
}

@MainActor
final class SessionManager {
    private let state = MutableStateFlow<SessionState>(.signedOut)
    var sessionState: any StateFlow<SessionState> { state }

    func signIn(email: String, password: String) async {
        await state.send(.signingIn)
        do {
            let user = try await authAPI.signIn(email: email, password: password)
            await state.send(.signedIn(user))
        } catch {
            await state.send(.error(error.localizedDescription))
        }
    }
}

// Multiple screens observe the same session state:
for await state in sessionManager.sessionState.asFlow() {
    switch state {
    case .signedIn(let user): updateProfile(user)
    case .signedOut: navigateToSignIn()
    case .signingIn: showLoadingIndicator()
    case .error(let message): showError(message)
    }
}

enum NavigationEvent: Sendable {
    case presentAlert(title: String, message: String)
    case navigateToDetail(productID: String)
    case dismissCurrentScreen
}

@MainActor
final class NavigationCoordinator {
    private let events = MutableSharedFlow<NavigationEvent>(replay: 0)
    var eventStream: any SharedFlow<NavigationEvent> { events }

    func present(alert title: String, message: String) async {
        await events.emit(.presentAlert(title: title, message: message))
    }
}

// In a root coordinator:
for await event in navigationCoordinator.eventStream.asFlow() {
    switch event {
    case .presentAlert(let title, let message):
        showAlert(title: title, message: message)
    case .navigateToDetail(let productID):
        navigator.push(ProductDetailScreen(productID: productID))
    case .dismissCurrentScreen:
        navigator.pop()
    }
}

probing reads a flow with no timeout. Each read waits for the flow's next emission, completion or failure, so a slow CI machine cannot fail a correct test.

import Testing
import Flow
import FlowTesting

@Test("article feed emits cached then fresh")
func articleFeedEmitsCachedThenFresh() async throws {
    let feed = ThrowingFlow<[Article]> { collector in
        try await collector.emit([.cached])
        try await collector.emit([.cached, .fresh])
    }

    try await feed.probing { reader in
        try await reader.expectValue([.cached])
        try await reader.expectValue([.cached, .fresh])
        try await reader.expectCompletion()
    }
}

test(timeout:_:), TestScope.run(timeout:_:) and waitUntil(timeout:_:) measure their deadline in real elapsed time and are deprecated in favour of probing. When a test does need a deadline, pass a TestClock and move time by hand: test(timeout:clock:_:), TestScope.run(timeout:clock:_:), withThrowingTimeout(_:clock:_:) and waitUntil(timeout:clock:_:) expire only when the clock is advanced.

TestClock.waitForSleepers(_:) suspends until a time-based operator has registered its sleep, and FlowProbe.waitForValue(where:) suspends until a tapped value has passed through the operator. MutableSharedFlow and MutableStateFlow have waitForSubscribers(_:) to wait for a collector to attach. Each has a falling counterpart, TestClock.waitForSleepers(atMost:) (and waitForNoSleepers()) and waitForSubscribers(atMost:), for waiting until a sleep is torn down or a collector detaches. None polls or uses a deadline, and each throws CancellationError if the waiting task is cancelled.

let clock = TestClock()
let probe = FlowProbe<Int>()
let sampled = upstream.asFlow().tap(after: probe).sample(every: .seconds(1), clock: clock)

try await sampled.probing { reader in
    try await upstream.waitForSubscribers(1)
    await upstream.emit(3)
    try await probe.waitForValue { $0 == 3 }
    try await clock.waitForSleepers(1)
    await clock.advance(by: .seconds(1))
    try await reader.expectValue(3)
}

To check that nothing else arrived, trigger a known emission and read for it with expectNextValue(_:) instead of sleeping. The one limit: a producer that stays open and never emits again cannot be told from a slow one without a clock, so use test(timeout:clock:_:) with a TestClock when you need that bound.

Flow<T> and ThrowingFlow<T> are cold async stream types. Each collector gets its own execution of the body closure. Builders include of:, sequence init, empty, never, and asFlow(). FlowScope handles lifetime, and launch(in:) ties collection to it.

StateFlow / MutableStateFlow give you a current value with built-in deduplication. SharedFlow / MutableSharedFlow broadcast to multiple subscribers with concurrent delivery and a configurable replay buffer. Convert cold to hot with asStateFlow(initialValue:strategy:) or asSharedFlow(replay:strategy:) using .eager, .lazy, or .whileSubscribed.

• Transform: map, compactMap, filter, transform, prefix, dropFirst, scan.
• Lifecycle: onStart, onEach, onCompletion.
• Flattening: flatMap, flatMap(maxConcurrent:), flatMapLatest.
• Combining: zip, combineLatest, merge.
• Error handling: catch, retry, retryWhen.
• Rate-limiting: debounce, throttle, removeDuplicates, sample.
• Buffering: buffer, keepingLatest.
• Terminal: collectLatest, first, exactlyOne, toArray, reduce.

import FlowUI gets you the SwiftUI @CollectedState property wrapper, ObservedStateFlow with isolated deinit, the View.collecting extension, and Flow(observing:_:) for @Observable types. On UIKit/AppKit, UIViewController.flowScope, NSViewController.flowScope, and NSWindowController.flowScope tie collection to view controller lifetimes.

FlowReader (via Flow.probing(_:)) reads a flow with no deadline. TestClock gives deterministic virtual time for rate-limiting and sharing operators, and FlowTester, ThrowingFlowTester and TestScope drive several flows at once against a deadline you measure on a TestClock.

If your change touches anything in Sources/Core/Flow/Documentation.docc/, regenerate the static site before opening a PR:

rm -rf docs
swift package --allow-writing-to-directory ./docs \
    generate-documentation --target Flow \
    --disable-indexing \
    --transform-for-static-hosting \
    --hosting-base-path FlowKit \
    --output-path ./docs

Commit both the source change and the regenerated docs/ directory in the same PR.

MIT. See LICENSE.