A Swift class designed for injecting dependency services with separate live/preview/test implementations into Swift apps using the Observation framework.

StatefulService.swift

import Observation

@Observable class StatefulService<State, API> {
    typealias Stream = StateStream<State>
    typealias StateStreamTask = Task<Void, Never>
    typealias OnStateChange = (State) -> Void
    
    /// The API of the service
    let api: API
    /// The state of the service
    private(set) var state: State
    /// A function that's called on each new state from the state stream
    private var onStateChange: OnStateChange?
    /// The cancellable state stream observation task. When cancelled, this breaks the for-await-in loop over the stream
    private var stateStreamTask: StateStreamTask
    
    /// Initializes a new stateful service
    /// - Parameter state: The initial state of the service
    /// - Parameter stream: A state-stream of state updates
    /// - Parameter api: The service's API
    /// - Parameter onStateChange: An optional closure that's called on each state change, which can be used to handle states
    required init(state: State, stream: Stream, api: API, onStateChange: OnStateChange? = nil) {
        self.api = api
        self.state = state
        self.onStateChange = onStateChange
        
        // Initialize stateStreamTask to a temporary value before creating the real task so the compiler
        // doesn't complain about self.stateStreamTask being captured before it was initialized
        self.stateStreamTask = Task {}
        self.stateStreamTask = observeStateStream(stream: stream)
    }
    
    deinit {
        stateStreamTask.cancel()
    }
    
    private func observeStateStream(stream: StateStream<State>) -> StateStreamTask {
        return Task {
            for await state in stream.stream {
                // If the state stream task is cancelled, break out of the loop, terminating the stream
                guard !Task.isCancelled else { break }
                // Update the state
                self.state = state
                // If an onStateChange handler is provided, call it with the new state
                if let onStateChange {
                    onStateChange(state)
                }
            }
        }
    }
}

struct StateStream<T: Sendable> {
    typealias Stream = AsyncStream<T>
    typealias Send = (T) -> Void
    
    private(set) var stream: Stream
    private(set) var send: Send!
    
    init(
        bufferingPolicy: Stream.Continuation.BufferingPolicy = .unbounded,
        body: @escaping (Stream.Continuation) -> Void
    ) {
        // Initialize stream to a temporary value before creating the real stream so the compiler
        // doesn't complain about self.stream being captured before it was initialized
        self.stream = Stream { _ in }
        
        // Construct the real stream
        self.stream = Stream(bufferingPolicy: bufferingPolicy) { (continuation: Stream.Continuation) in
            body(continuation)
            self.send = { (value: T) in
                continuation.yield(value)
            }
        }
    }
    
    /// Returns a void stream that never yields
    static var void: StateStream<()> {
        StateStream<()> { _ in }
    }
}