Flux

A lightweight, strongly-typed reactive programming library for Swift built on Swift Concurrency (AsyncStream, Task, Sendable) with zero external dependencies or runtime overhead.

Designed for modern Swift development, fully compliant with Swift 6 strict concurrency (-strict-concurrency=complete, Swift 5.9+ / Swift 6).

---

Key Features

• Swift 6 strict concurrency by design. Not retrofitted with @unchecked Sendable bandaids — every public type is built to compile clean under -strict-concurrency=complete from the ground up.
• Native Swift Concurrency, not a Combine wrapper. Built directly on AsyncStream / Task / Sendable — no bridging layer, no dependency on Combine (unavailable on Linux, increasingly legacy on Apple platforms).
• Explicit backpressure everywhere it matters. merge, combineLatest, flatMap, and Pipe all require an explicit BufferingPolicy. There is no silent, unbounded default that can grow memory under a slow consumer.
• Cancellation without leaks or races. TerminationStorage guarantees cleanup handlers still fire with the correct terminal reason even when cancellation happens mid-await (e.g. during a socket handshake) — a case most hand-rolled reactive wrappers get wrong.
• Race-free nested task management. TaskBox atomically replaces and cancels inner tasks (debounce timers, timeout watchdogs, flatMapLatest inner streams), so there's no window where a stale task can deliver after cancellation.
• Deterministic, RAII-style lifecycle. SubscriptionBag cancels everything automatically on deinit; no manual bookkeeping, no dangling subscriptions.
• Explicit cold/hot separation. Flux<T> is cold by default; Pipe, CurrentValue/CurrentValueDistinct, and Once are distinct, clearly-scoped hot primitives — no ambiguity like Combine's PassthroughSubject vs CurrentValueSubject.
• Ergonomic push syntax. The <- operator (events <- "value", events <- [a, b, c]) makes hot-source emission read naturally without sacrificing the explicit .send() API.
• Zero external dependencies. Nothing to audit beyond the Swift standard library and Concurrency runtime — important for security-sensitive networking code.
• Protocol-boundary discipline built in. A documented invariant keeps raw transport frames out of concurrent fan-in operators, preserving strict ordering where it's required (e.g. decrypt → decode → validate → publish).

---

Table of Contents

• Key Features
• Installation
• Architectural Concepts
• Quick Start
• Core Types
  • Flux<T> (Cold Stream)
  • Subscription and SubscriptionBag
  • FluxEmitter and Safe Cleanup
• State and Event Primitives
  • Pipe<T> (Hot Push Source)
  • CurrentValue<T> and CurrentValueDistinct<T> (State with Replay)
  • Once<T> (One-shot Promise)
• Creating Streams
• Operator Reference
  • Transformation
  • Filtering
  • Timing & Rate Limiting
  • Combining Multiple Streams
  • Dispatching & Side Effects
• Native Swift Async & Bridges
• Order Guarantees
• Protocol Boundary
• Practical Examples
  • Search-as-you-type
  • Auth Form Validation
  • Connection State UI Binding
• Building and Testing
• Changelog
• License

---

Installation

Swift Package Manager (SPM)

Add Flux to your Package.swift:

dependencies: [
    .package(url: "https://github.com/resoul/flux.git", from: "1.0.0")
]

Then add Flux to your target's dependencies:

.target(
    name: "YourApp",
    dependencies: [
        .product(name: "Flux", package: "flux")
    ]
)

Or add via Xcode: File → Add Package Dependencies... and enter https://github.com/resoul/flux.git.

Bazel

Add the library to your BUILD file:

swift_library(
    name = "MyTarget",
    srcs = glob(["Sources/**/*.swift"]),
    deps = [
        "@flux//:Flux",  # When imported as external repository
        # or "//submodules/Flux:Flux" if used as a submodule
    ],
)

---

Architectural Concepts

1. Cold by default: Flux<T> is a cold reactive stream. Each subscription (.sink(), .sinkOnMain(), or accessing .stream) invokes the producer closure anew and creates an independent AsyncStream.
2. Complete Sendable Safety: All types are designed for Swift 6 Complete Concurrency — zero unprotected shared mutable state (synchronization via NSLock / actor).
3. Deterministic Lifecycle: Subscriptions are managed explicitly via Subscription and SubscriptionBag (RAII pattern: all tasks are cancelled automatically upon container deinit).
4. Leak and Race Prevention on Cancellation:
   • TerminationStorage ensures cleanup handlers receive the terminal reason even if cancellation occurs during asynchronous resource initialization (e.g. while awaiting a socket connection).
   • TaskBox guarantees atomic replacement and cancellation of nested background tasks (watchdog in timeout, debounce delays, inner streams in flatMapLatest).
5. Explicit Backpressure / Buffering Policy: Multi-source fan-in operators (merge, combineLatest, flatMap) and hot sources (Pipe) require an explicit BufferingPolicy to protect memory from unbounded growth under slow downstream consumers.

---

Quick Start

import Flux

// 1. Create a stream pipeline
let searchFlux = Flux.from(["a", "ap", "app", "apple"])
    .debounce(.milliseconds(300))
    .flatMapLatest { query in
        api.search(query) // returns Flux<[SearchResult]>
    }

// 2. Subscribe on MainActor
let bag = SubscriptionBag()

searchFlux
    .sinkOnMain { results in
        tableView.reloadData(results)
    }
    .store(in: bag)

---

Core Types

Flux<T> (Cold Stream)

Flux<T: Sendable> is the fundamental struct of the library, wrapping an () -> AsyncStream<T> factory.

• Callback-based subscription:

  let sub = flux.sink { value in
      print("Received: \(value)")
  }
  
  // With completion handler:
  let subWithCompletion = flux.sink(
      next: { value in print(value) },
      completed: { print("Stream completed") }
  )

• UI Subscription:

  // Dispatched directly to @MainActor without redundant intermediate streams
  flux.sinkOnMain { [weak self] value in
      self?.label.stringValue = value
  }.store(in: bag)

Subscription and SubscriptionBag

• Subscription: Thread-safe handle for an active subscription. Calling sub.cancel() immediately stops delivery and cancels associated background tasks.
• SubscriptionBag: Subscription container. Automatically cancels all stored subscriptions on deinit or bag.cancelAll().

final class ChatViewController: NSViewController {
    private let bag = SubscriptionBag()
    
    override func viewDidLoad() {
        super.viewDidLoad()
        
        chatService.messages.flux
            .sinkOnMain { [weak self] message in
                self?.appendMessage(message)
            }
            .store(in: bag)
    }
    // When ChatViewController deinits, all subscriptions are automatically cancelled
}

FluxEmitter and Safe Cleanup

Use FluxEmitter<T> to construct custom Flux producers:

let networkStream = Flux<Data> { emitter in
    let socket = openSocket()
    
    // Register resource cleanup handlers
    emitter.onCancellation {
        socket.close() // called only when subscription is cancelled
    }
    
    emitter.onFinishOrCancel {
        // called on both normal finish and cancellation
    }
    
    socket.onData { data in
        emitter.send(data)
    }
    
    socket.onEnd {
        emitter.finish()
    }
}

Note: FluxEmitter uses TerminationStorage. If you register a cleanup handler after an await point (for instance, after awaiting a connection handshake) and the subscription was cancelled in the meantime, the cleanup callback is invoked immediately with the recorded termination reason.

---

State and Event Primitives

Pipe<T> (Hot Push Source)

A hot multicast broadcast source. Subscribers only receive values emitted after they subscribe.

let events = Pipe<String>(bufferingPolicy: .bufferingNewest(64))

// Subscribe
events.flux.sink { print("Event: \($0)") }.store(in: bag)

// Send via method:
events.send("Message 1")

// Send via operator <- :
events <- "Message 2"
events <- ["Batch 1", "Batch 2"]

// Buffer overflow telemetry:
let results = events.sendObservingOverflow("Critical Event")
if results.contains(where: { if case .dropped = $0 { return true }; return false }) {
    // Handle buffer overflow caused by slow subscriber
}

// Complete the pipe
events.finish()

CurrentValue<T> and CurrentValueDistinct<T> (State with Replay)

State holder actor.

• New subscribers immediately receive the current value (replay = 1), followed by all subsequent state updates.
• Default buffering policy: .bufferingNewest(1) (for a slow consumer, only the latest state matters).

enum ConnectionState: Sendable {
    case disconnected, connecting, connected
}

let state = CurrentValue<ConnectionState>(.disconnected)

// Read current value:
let current = await state.value

// Update state:
await state.set(.connecting)
await state.modify { _ in .connected }

// UI subscription:
state.flux
    .sinkOnMain { status in
        statusLabel.stringValue = "\(status)"
    }
    .store(in: bag)

CurrentValueDistinct<T: Equatable> behaves identically, but automatically deduplicates consecutive identical values (set(x) does not notify if the current value is already x).

Once<T> (One-shot Promise)

An actor holding a single asynchronous result.

let authReady = Once<Bool>()

// In an async initialization task:
await authReady.resolve(true)

// Await directly:
let isReady = await authReady.wait()

// Or reactive subscription (emits 1 value and completes):
authReady.flux.sink { isReady in ... }

---

Creating Streams

Method	Description
Flux.just(value)	Emits a single value and completes immediately.
Flux.empty()	Completes immediately without emitting any values.
Flux.never()	A stream that never emits values and never completes.
Flux.from(sequence)	Emits all elements from a collection/sequence in order, then completes.
Flux.timer(duration)	Waits for the specified Duration and completes without emitting values.
Flux { emitter in ... }	Custom generator closure with full control over emission and lifecycle callbacks.

---

Operator Reference

Transformation

• map { ... } — Synchronously transforms upstream values.
• asyncMap { await ... } — Asynchronously transforms values with await and Task.isCancelled checks.
• compactMap { ... } — Transforms values and unwraps/filters out nil results.
• flatMap(maxConcurrent:bufferingPolicy:) { ... } — Transforms each value into an inner Flux and merges results concurrently up to maxConcurrent active tasks.
• flatMapLatest { ... } — Transforms each value into an inner Flux, cancelling previous inner streams when a new outer value arrives.
• then(nextFlux) — Subscribes to nextFlux strictly after the current stream completes.

Filtering

• filter { predicate } — Forwards only values matching the predicate.
• take(count) — Emits the first count values and completes.
• skip(count) — Drops the first count values.
• prefix(while:) — Emits values as long as the predicate holds true, then completes.
• drop(while:) — Drops values as long as the predicate holds true, then emits all remaining values.
• skipRepeats() — Drops consecutive duplicate values (requires T: Equatable).

Timing & Rate Limiting

• delay(duration) — Delays delivery of each value by duration.
• debounce(duration) — Waits for a quiet period of duration before delivering the latest value (ideal for text search fields).
• throttle(duration) — Limits emission rate to at most once per duration (leading throttle — delivers the first value in each window).
• timeout(duration) — Completes the stream if no values are received within duration.

Combining Multiple Streams

• merge([flux1, flux2], bufferingPolicy:) — Merges values from multiple streams of the same type as they arrive.
• combineLatest(fa, fb, bufferingPolicy:) — Combines latest values from 2, 3, or 4 streams into a tuple whenever any input changes (starts emitting once each source has emitted at least once).
• zip(fa, fb) — Pairs values from two streams 1-to-1 (waits for (a1, b1), then (a2, b2)).

Dispatching & Side Effects

• onMain() — Dispatches downstream value delivery to MainActor.
• sinkOnMain { ... } — Directly subscribes on MainActor without intermediate stream allocations (recommended for UI).
• onBackground(priority:) — Dispatches delivery inside a detached background task with the specified priority.
• handleEvents { ... } — Executes a side effect for each value without mutating the stream (logging/metrics).
• onCompletion { ... } — Executes a side effect upon stream completion.

---

Native Swift Async & Bridges

Flux integrates seamlessly with standard Swift Concurrency:

// 1. Iterate using for await
for await value in flux.stream {
    print(value)
}

// 2. Fetch first value
if let first = await flux.first() {
    print("First value: \(first)")
}

// 3. Collect all values of a finite stream into an array
let allValues = await flux.collect()

// 4. Convert native AsyncStream into Flux
let fluxFromStream = nativeAsyncStream.asFlux()

// 5. Convert to throwing stream
let throwing = flux.throwingStream

---

Order Guarantees

Category	Operators	Order Guarantee
Sequential	map, filter, compactMap, take, skip, prefix, drop, then, skipRepeats	Result strictly follows upstream order.
Time-filtering	debounce, throttle	Relative order of delivered values is preserved (intermediate items may be dropped).
Concurrent (Fan-in / Multi-source)	merge, flatMap, combineLatest	Order between sources is not guaranteed (depends on parallel task completion and Swift Concurrency runtime scheduling).
Switch-to-latest	flatMapLatest	Previous inner task is guaranteed to be cancelled when a new outer value arrives.

---

Protocol Boundary

Important

Network Transport Invariant:

Flux is strictly intended for use after protocol decoding and validation — once raw network frames have been decrypted, validated, and converted into domain events (IncomingEvent).

The read loop remains strictly sequential: $$\text{read frame} \longrightarrow \text{decrypt / verify} \longrightarrow \text{decode} \longrightarrow \text{validate} \longrightarrow \text{demux} \longrightarrow \text{publish to Flux}$$

Raw frames should not pass through concurrent fan-in operators (merge, flatMap, combineLatest) to preserve frame ordering and transport context.

---

Practical Examples

1. Search-as-you-type

final class SearchViewModel {
    let queryPipe = Pipe<String>()
    private let bag = SubscriptionBag()
    
    @MainActor var searchResults: [SearchResult] = []
    
    init(api: SearchAPI) {
        queryPipe.flux
            .map { $0.trimmingCharacters(in: .whitespacesAndNewlines) }
            .filter { $0.count >= 2 }
            .debounce(.milliseconds(300))
            .skipRepeats()
            .flatMapLatest { query in
                api.searchFlux(query) // cancels previous in-flight network requests on new input
            }
            .sinkOnMain { [weak self] results in
                self?.searchResults = results
            }
            .store(in: bag)
    }
}

2. Auth Form Validation

let email = CurrentValue("")
let password = CurrentValue("")

let isFormValid = combineLatest(
    email.flux,
    password.flux,
    bufferingPolicy: .bufferingNewest(1)
).map { email, pass in
    email.contains("@") && pass.count >= 8
}

isFormValid
    .sinkOnMain { isValid in
        loginButton.isEnabled = isValid
    }
    .store(in: bag)

3. Connection State UI Binding

client.connectionState?.sinkOnMain { [weak self] state in
    switch state {
    case .connected:
        self?.statusIndicator.color = .systemGreen
    case .reconnecting(let attempt):
        self?.statusIndicator.color = .systemOrange
    case .disconnected:
        self?.statusIndicator.color = .systemRed
    }
}.store(in: bag)

---

Building and Testing

Running Tests

# Swift Package Manager:
swift test

# Bazel:
bazel test //:FluxTests

---

Changelog

See CHANGELOG.md for detailed release history and changes.

---

License

Flux is released under the MIT License.