EffectsHandler: (Publisher<Effect>) -> Publisher<ReAction>

CombineExtensions.swift

import Combine

extension Publisher {
    
    public func flatMap<T, P, UP>(
        maxPublishers: Subscribers.Demand = .unlimited,
        transform: @escaping (Self.Output) -> P,
        until: @autoclosure @escaping () -> UP
    ) -> Publishers.FlatMap<Publishers.PrefixUntilOutput<P, UP>, Self>
        where T == P.Output, P: Publisher, Self.Failure == P.Failure, UP: Publisher {
        flatMap(maxPublishers: maxPublishers) { output in
            transform(output)
                .prefix(untilOutputFrom: until())
        }
    }
    
    public func flatMapIgnoringOthers<T, P>(
        maxPublishers: Subscribers.Demand = .unlimited,
        debug: Bool = false,
        _ transform: @escaping (Self.Output) -> P
    ) -> Publishers.HandleEvents<Publishers.FlatMap<P, Publishers.HandleEvents<Publishers.Filter<Self>>>>
        where T == P.Output, P: Publisher, Self.Failure == P.Failure {
        var permitted = true {
            didSet {
                if debug {
                    Swift.print("    permitted", permitted)
                }
            }
        }
        return self
            .filter { _ in
                if debug, !permitted {
                    Swift.print("    skip")
                }
                return permitted
            }
            .handleEvents(receiveOutput: { _ in permitted = false })
            .flatMap(maxPublishers: maxPublishers, transform)
            .handleEvents(receiveOutput: { _ in permitted = true })
    }
    
    public func compactMap<T>(_ keyPath: KeyPath<Self.Output, T?>) -> Publishers.CompactMap<Self, T> {
        compactMap({ $0[keyPath: keyPath] })
    }
}

EffectsHandler.swift

import Combine

public struct AnyEffectsHandler<Effect, Action> {
    private let _handle: (AnyPublisher<Effect, Never>) -> AnyPublisher<Action, Never>
    
    public init<ActionsPublisher: Publisher>(handle: @escaping (AnyPublisher<Effect, Never>) -> ActionsPublisher) where
        ActionsPublisher.Output == Action,
        ActionsPublisher.Failure == Never {
        self._handle = { handle($0).eraseToAnyPublisher() }
    }
    
    public func handle<EffectsPublisher: Publisher>(_ effects: EffectsPublisher) -> AnyPublisher<Action, Never> where
        EffectsPublisher.Output == Effect,
        EffectsPublisher.Failure == Never {
        _handle(effects.eraseToAnyPublisher())
    }
}

extension AnyEffectsHandler {
    
    public init(
        _ a: AnyEffectsHandler<Effect, Action>,
        _ b: AnyEffectsHandler<Effect, Action>,
        _ c: AnyEffectsHandler<Effect, Action>
    ) {
        self.init(handle: { effects in
            Publishers.Merge3(
                a.handle(effects),
                b.handle(effects),
                c.handle(effects)
            )
        })
    }
    
    public static func simple(block: @escaping (Effect, _ onUpdate: @escaping (Action) -> Void) -> Void) -> AnyEffectsHandler<Effect, Action> {
        AnyEffectsHandler(handle: { effects in
            effects
                .flatMap { effect -> PassthroughSubject<Action, Never> in
                    let actionSubject = PassthroughSubject<Action, Never>()
                    block(effect, actionSubject.send)
                    return actionSubject
                }
        })
    }
    
    public func map<HigherAction>(
        action toHigherAction: @escaping (Action) -> HigherAction
    ) -> AnyEffectsHandler<Effect, HigherAction> {
        AnyEffectsHandler<Effect, HigherAction>(handle: { effects in
            self.handle(effects).map(toHigherAction)
        })
    }
    
    public func pullback<HigherEffect>(
        effect toLowerEffect: @escaping (HigherEffect) -> Effect?
    ) -> AnyEffectsHandler<HigherEffect, Action> {
        AnyEffectsHandler<HigherEffect, Action>(handle: { effects in
            self.handle(effects.compactMap(toLowerEffect))
        })
    }
    
}

example.swift

import Combine
import Foundation

private enum IncomingEffect {
    case fetchBoard(String, String)
}

extension IncomingEffect {
    /// Keypath helper
    var fetchBoard: (session: String, boardID: String)? {
        guard case .fetchBoard(let value) = self else { return nil }
        return value
    }
}

private enum OutgoingAction {
    case boardUpdated(String)
}

private func fetchBoard(session: String, boardID: String) -> Future<Result<String, Error>, Never> {
    print("  🛫", #function, boardID)
    return Future({ completion in
        DispatchQueue.main.asyncAfter(deadline: .now() + 0.2) {
            completion(.success(.success("new board: \(boardID)")))
        }
    })
}

private final class MiniBoardFetcher {
    let session: String
    init(session: String) { self.session = session }
    func fetch(boardID: String) -> Future<Result<String, Error>, Never> {
        fetchBoard(session: session, boardID: boardID)
    }
}

// MARK: - Examples

private let cancelAndTakeLatest = AnyEffectsHandler<IncomingEffect, OutgoingAction> { effects in
    effects
        .compactMap(\.fetchBoard)
        .flatMap(
            transform: fetchBoard,
            until: effects.compactMap(\.fetchBoard)
        )
        .compactMap { try? $0.get() }
        .map(OutgoingAction.boardUpdated)
}

private let waitFor1SecondBeforeFiringFirstRequest = AnyEffectsHandler<IncomingEffect, OutgoingAction> { effects in
    effects
        .compactMap(\.fetchBoard)
        .throttle(for: .milliseconds(1000), scheduler: RunLoop.main, latest: false)
        .flatMap(fetchBoard)
        .compactMap { try? $0.get() }
        .map(OutgoingAction.boardUpdated)
}

private let ignoreNewRequestsWhileInFlight = AnyEffectsHandler<IncomingEffect, OutgoingAction> { effects in
    effects
        .compactMap(\.fetchBoard)
        .flatMapIgnoringOthers(fetchBoard)
        .compactMap { try? $0.get() }
        .map(OutgoingAction.boardUpdated)
}

private let hypotheticalMerged = AnyEffectsHandler<IncomingEffect, OutgoingAction>(
    cancelAndTakeLatest,
    waitFor1SecondBeforeFiringFirstRequest,
    ignoreNewRequestsWhileInFlight
)

private func injectionYo(
    fetcher: @escaping (_ session: String) -> MiniBoardFetcher
) -> AnyEffectsHandler<IncomingEffect, OutgoingAction> {
    AnyEffectsHandler<IncomingEffect, OutgoingAction> { effects in
        effects
            .compactMap(\.fetchBoard)
            .flatMap { effect in
                fetcher(effect.0)
                    .fetch(boardID: effect.1)
            }
            .compactMap { try? $0.get() }
            .map(OutgoingAction.boardUpdated)
    }
}

private var neverDispose: Set<AnyCancellable> = []
private let simpleMode = AnyEffectsHandler<IncomingEffect, OutgoingAction>.simple { (effect, onUpdate) in
    switch effect {
    case .fetchBoard(let session, let boardID):
        fetchBoard(session: session, boardID: boardID)
            .sink(receiveValue: { resultOfBoard in
                switch resultOfBoard {
                case .success(let board):
                    onUpdate(.boardUpdated(board))
                case .failure(let error):
                    print("fetch board failed", error)
                }
            })
            .store(in: &neverDispose) // Our BrightFutures are self-retaining
        // case ...
    }
}

// MARK: - Setup -

private let effects = PassthroughSubject<IncomingEffect, Never>()
var disposebag: Set<AnyCancellable> = []

private func effectStream3() {
    cancelAndTakeLatest.handle(effects)
        .sink(receiveValue: { print("cancelAndTakeLatest()", $0) })
        .store(in: &disposebag)

    waitFor1SecondBeforeFiringFirstRequest.handle(effects)
        .sink(receiveValue: { print("waitFor1SecondBeforeFiringFirstRequest()", $0) })
        .store(in: &disposebag)

    ignoreNewRequestsWhileInFlight.handle(effects)
        .sink(receiveValue: { print("ignoreNewRequestsWhileInFlight()", $0) })
        .store(in: &disposebag)

    hypotheticalMerged.handle(effects)
        .sink(receiveValue: { print("hypotheticalMerged()", $0) })
        .store(in: &disposebag)

    simpleMode.handle(effects)
        .sink(receiveValue: { print("simpleMode()", $0) })
        .store(in: &disposebag)
    
    var globalStoreOfBoardFether: [String: MiniBoardFetcher] = [:]
    injectionYo(
        fetcher: { session in
            if let cachedFetcher = globalStoreOfBoardFether[session] {
                return cachedFetcher
            } else {
                let newFetcher = MiniBoardFetcher(session: session)
                globalStoreOfBoardFether[session] = newFetcher
                return newFetcher
            }
        })
        .handle(effects)
        .sink(receiveValue: { print("injectionYo()", $0) })
        .store(in: &disposebag)

    // MARK: - RUN -
    
    print("r1-0")
    effects.send(.fetchBoard("session", "board1+r1"))
    
    print("r2-0")
    effects.send(.fetchBoard("session", "board1+r2"))
    
    DispatchQueue.main.async {
        print("r3+0")
        effects.send(.fetchBoard("session", "board1+r3"))
    }
    
    DispatchQueue.main.asyncAfter(deadline: .now() + 0.3) {
        print("r4+.3")
        effects.send(.fetchBoard("session", "board1+r4"))
    }
    DispatchQueue.main.asyncAfter(deadline: .now() + 0.4) {
        print("r5+.4")
        effects.send(.fetchBoard("session", "board1+r5"))
    }
    DispatchQueue.main.asyncAfter(deadline: .now() + 0.5) {
        print("r6+.5")
        effects.send(.fetchBoard("session", "board1+r6"))
    }
    DispatchQueue.main.asyncAfter(deadline: .now() + 0.6) {
        print("r7+.6")
        effects.send(.fetchBoard("session", "board1+r7"))
        print("r8+.6")
        effects.send(.fetchBoard("session", "board1+r8"))
        print("r9+.6")
        effects.send(.fetchBoard("session", "board1+r9"))
    }
}

effectStream3()