Invert subscription transforms in a reactive way (attempt No. 2)

ReSwift/CoreTypes/Inversion/Cancelable.swift

@@ -0,0 +1,12 @@
+//
+//  Cancelable.swift
+//  ReSwift
+//
+//  Created by Christian Tietze on 2019-08-03.
+//  Copyright © 2019 ReSwift. All rights reserved.
+//
+
+/// Intended for side-effects.
+protocol Cancelable: Disposable {
+    var isDisposed: Bool { get }
+}

ReSwift/CoreTypes/Inversion/Disposable.swift

@@ -0,0 +1,47 @@
+//
+//  Disposable.swift
+//  ReSwift
+//
+//  Created by Christian Tietze on 2019-08-03.
+//  Copyright © 2019 ReSwift. All rights reserved.
+//
+
+internal protocol Disposable {
+    /// Dispose resource callback.
+    func dispose()
+}
+
+/// Create disposable that does nothing when cleaning up resources.
+func createDisposable() -> Disposable {
+    return NullDisposable.noOp
+}
+
+func createDisposable(with action: @escaping () -> Void) -> Disposable {
+    return AnonymousDisposable(action: action)
+}
+
+private struct NullDisposable: Disposable {
+    fileprivate static let noOp: Disposable = NullDisposable()
+
+    init() {}
+
+    func dispose() {
+        // no-op
+    }
+}
+
+private final class AnonymousDisposable: Disposable {
+    public typealias DisposeAction = () -> Void
+
+    private var disposeAction: DisposeAction?
+
+    init(action: @escaping DisposeAction) {
+        self.disposeAction = action
+    }
+
+    func dispose() {
+        guard let action = self.disposeAction else { return }
+        self.disposeAction = nil
+        action()
+    }
+}

ReSwift/CoreTypes/Inversion/IncompleteSubscription.swift

@@ -0,0 +1,57 @@
+//
+//  IncompleteSubscription.swift
+//  ReSwift
+//
+//  Created by Christian Tietze on 2019-08-04.
+//  Copyright © 2019 ReSwift. All rights reserved.
+//
+
+internal final class BlockSubscriber<S>: StoreSubscriber {
+    typealias StoreSubscriberStateType = S
+    private let block: (S) -> Void
+
+    init(block: @escaping (S) -> Void) {
+        self.block = block
+    }
+
+    func newState(state: S) {
+        self.block(state)
+    }
+}
+
+public final class IncompleteSubscription<RootStoreState: StateType, Substate> {
+    typealias CompatibleStore = Store<RootStoreState>
+
+    internal let store: CompatibleStore
+    internal let observable: Observable<Substate>
+
+    /// Used during transformations.
+    internal init(store: CompatibleStore, observable: Observable<Substate>) {
+        self.store = store
+        self.observable = observable
+    }
+
+    func asObservable() -> Observable<Substate> {
+        return observable
+    }
+}
+
+extension IncompleteSubscription {
+    @discardableResult
+    public func subscribe<Subscriber: StoreSubscriber>(_ subscriber: Subscriber)
+        -> SubscriptionToken
+        where Subscriber.StoreSubscriberStateType == Substate
+    {
+        return self.store.subscribe(subscription: self, subscriber: subscriber)
+    }
+}
+
+extension IncompleteSubscription where Substate: Equatable {
+    @discardableResult
+    public func subscribe<Subscriber: StoreSubscriber>(_ subscriber: Subscriber)
+        -> SubscriptionToken
+        where Subscriber.StoreSubscriberStateType == Substate
+    {
+        return self.store.subscribe(subscription: self, subscriber: subscriber)
+    }
+}

ReSwift/CoreTypes/Inversion/Observable+create.swift

@@ -0,0 +1,70 @@
+//
+//  Observable+create.swift
+//  ReSwift
+//
+//  Created by Christian Tietze on 2019-08-03.
+//  Copyright © 2019 ReSwift. All rights reserved.
+//
+
+extension Observable {
+    /// Starting point of events, aka state updates.
+    ///
+    /// The `producer` block is first invoked when an actual connection is made.
+    /// Until you finish the subscription, no callbacks are being made.
+    internal static func create(_ producer: @escaping (AnyObserver<Substate>) -> Disposable) -> Observable<Substate> {
+        return ObservableEventSource(producer: producer)
+    }
+}
+
+/// The source of an event sequence that all operators will eventually delegate to.
+///
+/// Overrides the abstract `Producer.run` to actually connect the event creation sequence to an
+/// observer.
+///
+/// `ObservableEventSource.producer` is a closure that is configured upon creation to produce `.on`
+/// events; they are passed to the observer that is eventually set via `subscribe`.
+final private class ObservableEventSource<Substate>: Producer<Substate> {
+    typealias EventProducer = (_ consumer: AnyObserver<Substate>) -> Disposable
+
+    internal let producer: EventProducer
+
+    init(producer: @escaping EventProducer) {
+        self.producer = producer
+    }
+
+    override func run<Observer: ObserverType>(
+        _ observer: Observer,
+        cancel: Cancelable
+        ) -> (sink: Disposable, subscription: Disposable)
+        where Observer.Substate == Substate
+    {
+        let sink = ObservableEventSourceSink(observer: observer, cancel: cancel)
+        let subscription = sink.subscribeObserver(self)
+        return (sink, subscription)
+    }
+}
+
+/// In short, objects of this type are adapters between `ObservableEventSource.EventProducer`
+/// and an a compatible `ObserverType` that perform no additional actions; their sole
+/// reason to exist is memory management of the `observer` and `cancel` associated with `Sink`s.
+///
+/// Since this type is a `Sink`, it keeps a strong reference to both its `observer`
+/// and `cancel` handler. It will be released when the connection itself is released.
+/// The connection from event source to observer will be released when `cancel` is disposed,
+/// which ultimately is a `Producer`'s `SinkDisposer`.
+final private class ObservableEventSourceSink<Observer: ObserverType>: Sink<Observer>, ObserverType {
+    typealias Substate = Observer.Substate
+
+    override init(observer: Observer, cancel: Cancelable) {
+        super.init(observer: observer, cancel: cancel)
+    }
+
+    func on(_ state: Substate) {
+        self.observer.on(state)
+    }
+
+    func subscribeObserver(_ eventSource: ObservableEventSource<Substate>) -> Disposable {
+        let erasedObserver = AnyObserver(self)
+        return eventSource.producer(erasedObserver)
+    }
+}

ReSwift/CoreTypes/Inversion/Observable.swift

@@ -0,0 +1,53 @@
+//
+//  Observable.swift
+//  ReSwift
+//
+//  Created by Christian Tietze on 2019-08-03.
+//  Copyright © 2019 ReSwift. All rights reserved.
+//
+
+/// Represents a push style state update sequence.
+protocol ObservableType {
+    associatedtype Substate
+    func subscribe<Observer: ObserverType>(_ observer: Observer) -> Disposable where Observer.Substate == Substate
+}
+
+extension ObservableType {
+    func subscribe(_ observer: @escaping (Substate) -> Void) -> Disposable {
+        let observer = AnyObserver(observer: observer)
+        return self.asObservable().subscribe(observer)
+    }
+}
+
+extension ObservableType {
+    func asObservable() -> Observable<Substate> {
+        return Observable.create { observer in
+            return self.subscribe(observer)
+        }
+    }
+}
+
+/// Type-erased `ObervableType`.
+///
+/// You can only get `Observable` instances through the `create` static factory, as used by the `Store`.
+internal class Observable<Substate>: ObservableType {
+    internal init() {
+        // no-op
+    }
+
+    func subscribe<Observer: ObserverType>(_ observer: Observer) -> Disposable where Observer.Substate == Substate {
+        fatalError("Method not implemented: \(#function)")
+    }
+
+    /// Optimization hook for multiple `select` calls in succession. Is overwritten by the `Select` type.
+    internal func composeSelect<SelectedSubstate>(
+        _ transform: @escaping (Substate) -> SelectedSubstate
+        ) -> Observable<SelectedSubstate>
+    {
+        return ReSwift.select(source: self, transform: transform)
+    }
+
+    func asObservable() -> Observable<Substate> {
+        return self
+    }
+}

ReSwift/CoreTypes/Inversion/Observer.swift

@@ -0,0 +1,30 @@
+//
+//  Observer.swift
+//  ReSwift
+//
+//  Created by Christian Tietze on 2019-08-03.
+//  Copyright © 2019 ReSwift. All rights reserved.
+//
+
+/// Receives a state update.
+protocol ObserverType {
+    associatedtype Substate
+    func on(_ state: Substate)
+}
+
+/// Type-erased `ObserverType` that forwards events.
+internal final class AnyObserver<Substate>: ObserverType {
+    private let observer: (Substate) -> Void
+
+    init(observer: @escaping (Substate) -> Void) {
+        self.observer = observer
+    }
+
+    init<Observer: ObserverType>(_ observer: Observer) where Observer.Substate == Substate {
+        self.observer = observer.on
+    }
+
+    func on(_ substate: Substate) {
+        self.observer(substate)
+    }
+}

ReSwift/CoreTypes/Inversion/Operators/Filter.swift

@@ -0,0 +1,61 @@
+//
+//  Filter.swift
+//  ReSwift
+//
+//  Created by Christian Tietze on 2019-08-03.
+//  Copyright © 2019 ReSwift. All rights reserved.
+//
+
+extension IncompleteSubscription {
+    public func filter(_ predicate: @escaping (Substate) -> Bool) -> IncompleteSubscription<RootStoreState, Substate> {
+        return IncompleteSubscription<RootStoreState, Substate>(
+            store: self.store,
+            observable: self.observable.filter(predicate))
+    }
+}
+
+extension ObservableType {
+    func filter(_ predicate: @escaping (Substate) -> Bool) -> Observable<Substate> {
+        return Filter(source: self.asObservable(), predicate: predicate)
+    }
+}
+
+final private class Filter<Substate>: Producer<Substate> {
+    typealias Predicate = (Substate) -> Bool
+
+    private let source: Observable<Substate>
+    private let predicate: Predicate
+
+    init(source: Observable<Substate>, predicate: @escaping Predicate) {
+        self.source = source
+        self.predicate = predicate
+    }
+
+    override func run<Observer: ObserverType>(
+        _ observer: Observer,
+        cancel: Cancelable
+        ) -> (sink: Disposable, subscription: Disposable)
+        where Observer.Substate == Substate
+    {
+        let sink = FilterSink(predicate: self.predicate, observer: observer, cancel: cancel)
+        let subscription = self.source.subscribe(sink)
+        return (sink, subscription)
+    }
+}
+
+final private class FilterSink<Observer: ObserverType>: Sink<Observer>, ObserverType {
+    typealias Substate = Observer.Substate
+    typealias Predicate = (Substate) -> Bool
+
+    private let predicate: Predicate
+
+    init(predicate: @escaping Predicate, observer: Observer, cancel: Cancelable) {
+        self.predicate = predicate
+        super.init(observer: observer, cancel: cancel)
+    }
+
+    func on(_ state: Substate) {
+        guard self.predicate(state) else { return }
+        self.forward(state: state)
+    }
+}