Observing dependencies

Observing.swift

switch action {
case .task:
    return .observe {
        myDependency.events => Action.onEvent
    }
}

// where this is put at the end of the View:

.task {
    await viewStore.send(.task).finish()
}

// and the dependency exposes something like this

struct MyDependency {
    var events: AsyncPassthroughSubject<Event> = .init()
    
    func somethingHappened() {
        events.send(.somethingHappaned) 
    }
}

// AsyncPassthroughSubject is from AsyncExtensions, you can also do this using only Combine

struct MyDependency {
    private var _events = PassthroughSubject<Event, Never>()

    var events: AsyncPublisher<PassthroughSubject<Event, Never>> { _events.values }
    
    func somethingHappened() {
        _events.send(.somethingHappaned) 
    }
}


// MARK: The plumbing

precedencegroup EmbeddingPrecedence {
    higherThan: MultiplicationPrecedence
    associativity: right
}

infix operator =>: EmbeddingPrecedence

func =><S: AsyncSequence, Action>(sequence: S, action: (S.Element) -> Action) -> Observation<Action> {
    .init(sequence.eraseToStream(), action: action)
}

func =><S: AsyncSequence, Value, Action>(sequence: S, action: (Value) -> Action) -> Observation<Action> where S.Element == Value? {
    .init(sequence.compacted().eraseToStream(), action: action)
}

extension AsyncSequence where Element: Equatable {
    var unique: AsyncRemoveDuplicatesSequence<Self> {
        removeDuplicates()
    }
}

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

    func callAsFunction(send: Effect<Action>.Send) async -> Void {
        await closure(send)
    }

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

extension Effect where Failure == Never {
    static func observe<Value>(_ stream: AsyncStream<Value>, action: @escaping (Value) -> Action) -> Self {
        run { send in
            for await value in stream {
                await send(action(value))
            }
        }
    }

    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(send: send)
                    }
                }
            }
        }
    }
}