Observe Dependency

ObserveDependency.swift

struct TestReducer: Reducer {
    @Dependency(\.repo) var repo

    struct State: Equatable {
        var string: String = ""
        var int: Int = 0
    }

    enum Action {
        case task
        case onInt(Int)
        case onString(String)
    }

    func reduce(into state: inout State, action: Action) -> Effect<Action> {
        switch action {
            case .task:
                return .observe {
                    repo.strings => Action.onString
                    repo.ints => Action.onInt
                }

            case .onInt(let int):
                state.int = int
                return .none

            case .onString(let string):
                state.string = string
                return .none
        }
    }
}

// Where 

struct Repo {
    var ints: AsyncStream<Int> { ... }
    var strings: AsyncStream<String> { ... }
}

// Behind the scenes

@resultBuilder
enum ObservationBuilder<Action> {
    static func buildBlock(_ components: Observation<Action>...) -> [Observation<Action>] {
        components
    }
}

struct Observation<Action> {
    let f: (Effect<Action>.Send) async -> Void

    init<Value>(_ stream: AsyncStream<Value>, action: @escaping (Value) -> Action) {
        self.f = { send in
            for await value in stream {
                send(action(value))
            }
        }
    }
}

extension Effect where Failure == Never {
    static func observe(@ObservationBuilder<Action> _ observations: () -> [Observation<Action>]) -> Self {
        let obs = observations()
        return run { send in
            await withTaskGroup(of: Void.self) { group in
                for observation in obs {
                    group.addTask {
                        await observation.f(send)
                    }
                }
            }
        }
    }
}

infix operator =>

func =><Value, Action>(stream: AsyncStream<Value>, action: @escaping (Value) -> Action) -> Observation<Action> {
    .init(stream, action: action)
}