Rx state consolidation

Observable+consolidate.swift

import Foundation
import RxSwift

fileprivate final class MutableReference<T> {
    var value: T
    
    init(_ value: T) { self.value = value }
}

struct SideEffect<State> {
    fileprivate let observable: (MutableReference<State>, SchedulerType) -> Observable<Void>
    
    init<Observed>(_ observable: Observable<Observed>, _ update: @escaping (inout State, Observed) -> Void) {
        self.observable = { state, scheduler in
            return observable
                .observeOn(scheduler)
                .do(onNext: { value in
                    update(&state.value, value)
                })
                .map { _ in () }
        }
    }
}

extension Observable {
    
    func asSideEffect<State>(_ update: @escaping (inout State, Element) -> Void) -> SideEffect<State> {
        return SideEffect(self, update)
    }
    
    static func consolidate(_ initialValue: @escaping @autoclosure () -> Element,
                            with sideEffects: SideEffect<Element>...) -> Observable<Element> {
        return Observable.create { observer in
            let scheduler = SerialDispatchQueueScheduler(qos: .default)
            let state = MutableReference(initialValue())
            
            let sideEffectsStreams = sideEffects.map { $0.observable(state, scheduler) }
            return Observable<Observable<Void>>.from(sideEffectsStreams)
                .merge()
                .observeOn(scheduler)
                .map { _ in state.value }
                .subscribe(observer)
        }
    }
}