[Proposal] Introduce ObservableSystem make system chainable.

[beeth0ven]

Overview

We can introduce ObservableSystem, this will make system chainable:

PlayCatch Example

Before:

let bindUI: (ObservableSchedulerContext<State>) -> Observable<Event> = bind(self) { me, state in
    ...
    return Bindings(subscriptions: subscriptions, events: events)
}

Observable.system(
    initialState: State.humanHasIt,
    reduce: { (state: State, event: Event) -> State in
        switch event {
        case .throwToMachine:
            return .machineHasIt
        case .throwToHuman:
            return .humanHasIt
        }
    },
    scheduler: MainScheduler.instance,
    feedback:
        // UI is human feedback
        bindUI,
        // NoUI, machine feedback
        react(request: { $0.machinePitching }, effects: { (_) -> Observable<Event> in
            return Observable<Int>
                .timer(.seconds(1), scheduler: MainScheduler.instance)
                .map { _ in Event.throwToHuman }
        })
    )
    .subscribe()
    .disposed(by: disposeBag)

After:

ObservableSystem.create(
    initialState: State.humanHasIt,
    reduce: { (state: State, event: Event) -> State in
        switch event {
        case .throwToMachine:
            return .machineHasIt
        case .throwToHuman:
            return .humanHasIt
        }
    },
    scheduler: MainScheduler.instance
    )
    .binded(self) { me, state in
        ...
        return Bindings(subscriptions: subscriptions, events: events)
    }
    .reacted(request: { $0.machinePitching }, effects: { (_) -> Observable<Event> in
        return Observable<Int>
            .timer(.seconds(1), scheduler: MainScheduler.instance)
            .map { _ in Event.throwToHuman }
    })
    .system([])
    .subscribe()
    .disposed(by: disposeBag)

Evolution

The solution is inspired by Rx. Let's get in.

What do we have currently in Rx?

I will show minimal type inferface in Rx, as it will help us move fast to destination:

typealias Event<Element> = Element // mocked, just a name

typealias Observer<Element> = (Event<Element>) -> Void

typealias Disposable = () -> Void

typealias Observable<Element> = (@escaping Observer<Element>) -> Disposable

I've removed unrelate logic to make our evolution "pure".

Now we can adds some operators which are free functions:

func filter<Element>(
    _ predicate: @escaping (Element) -> Bool
    ) -> (@escaping Observable<Element>) -> Observable<Element> {

    return { source -> Observable<Element> in
        ...
    }
}

func map<Element, Result>(
    _ transform: @escaping (Element) -> Result
    ) -> (@escaping Observable<Element>) -> Observable<Result> { ... }

func flatMap<Element, Result>(
    _ transform: @escaping (Element) -> Observable<Result>
    ) -> (@escaping Observable<Element>) -> Observable<Result> { ... }

As far as we can tell, Operator behaiver like a Middleware:

typealias Middleware<Element, Result> = (@escaping Observable<Element>) -> Observable<Result>

We can change operator a little bit to:

func fulter1<Element>(_ predicate: @escaping (Element) -> Bool) -> Middleware<Element, Element> { ... }

func map1<Element, Result>(_ transform: @escaping (Element) -> Result) -> Middleware<Element, Result> { ... }

func flatMap1<Element, Result>(_ transform: @escaping (Element) -> Observable<Result>) -> Middleware<Element, Result> { ... }

That's what we have now in Rx.

Port to RxFeedback

We can find a way to port all these stuff to RxFeedback:

What do we have in RxFeedback?

typealias Feedback<State, Event> = (Observable<State>) -> Observable<Event>

typealias ImmediateSchedulerType = Any // Ignored in this demo context.

func system<State, Event>(
    initialState: State,
    reduce: @escaping (State, Event) -> State,
    scheduler: ImmediateSchedulerType,
    feedback: [Feedback<State, Event>]
    ) -> Observable<State> { ... }

We may add a createSystem function:

func createSystem<State, Event>(
    initialState: State,
    reduce: @escaping (State, Event) -> State,
    scheduler: ImmediateSchedulerType
    ) -> ([Feedback<State, Event>]) -> Observable<State> {
    
    return { feedback -> Observable<State> in
        ...
    }
}

By comparing function system with createSystem, It's not hard to find the return type has been changed form Observable<State> to ([Feedback<State, Event>]) -> Observable<State>.

Ok. This will open a new world, let's call the new return type System:

typealias System<State, Event> = ([Feedback<State, Event>]) -> Observable<State>

Then createSystem becomes:

func createSystem1<State, Event>(
    initialState: State,
    reduce: @escaping (State, Event) -> State,
    scheduler: ImmediateSchedulerType
    ) -> System<State, Event> { ... }

Next we can introduce SystemMiddleware:

typealias SystemMiddleware<State, Event> = (System<State, Event>) -> System<State, Event>

The feedback creator funtion like react and bind in RxFeedback now becomes operator:

func react<State, Request: Equatable, Event>(
    request: @escaping (State) -> Request?,
    effects: @escaping (Request) -> Observable<Event>
    ) -> SystemMiddleware<State, Event> { ... }

func react<State, Request: Equatable, Event>(
    requests: @escaping (State) -> Set<Request>,
    effects: @escaping (Request) -> Observable<Event>
    ) -> SystemMiddleware<State, Event> { ... }

func bind<State, Event>(
    _ bindings: @escaping (Observable<State>) -> (subscriptions: [Disposable], events: [Observable<Event>])
    ) -> SystemMiddleware<State, Event> { ... }

Real

Let's bring this to real.

Introduce ObservableSystem to RxFeedback:

public struct ObservableSystem<State, Event> {
    public typealias Feedback = Observable<Any>.Feedback<State, Event>
    public typealias System = ([Feedback]) -> Observable<State>
    
    public let system: System
    
    private init(_ system: @escaping System) {
        self.system = system
    }
}

extension ObservableSystem {
    
    public static func create(
        initialState: State,
        reduce: @escaping (State, Event) -> State,
        scheduler: ImmediateSchedulerType
        ) -> ObservableSystem<State, Event> {
        return ObservableSystem { feedback in
            return Observable<Any>.system(
                initialState: initialState,
                reduce: reduce,
                scheduler: scheduler,
                feedback: feedback
            )
        }
    }
    
    public func reacted<Request: Equatable>(
        request: @escaping (State) -> Request?,
        effects: @escaping (Request) -> Observable<Event>
        ) -> ObservableSystem<State, Event> {
        let newFeedback: Feedback = react(request: request, effects: effects)
        let sourceSystem = self.system
        return ObservableSystem { feedback in sourceSystem([newFeedback] + feedback) }
    }
    
    public func reacted<Request: Equatable>(
        requests: @escaping (State) -> Set<Request>,
        effects: @escaping (Request) -> Observable<Event>
        ) -> ObservableSystem<State, Event> {
        let newFeedback: Feedback = react(requests: requests, effects: effects)
        let sourceSystem = self.system
        return ObservableSystem { feedback in sourceSystem([newFeedback] + feedback) }
    }
    
    public func binded<WeakOwner: AnyObject>(
        _ owner: WeakOwner,
        _ bindings: @escaping (WeakOwner, ObservableSchedulerContext<State>) -> (Bindings<Event>)
        ) -> ObservableSystem<State, Event> {
        let newFeedback: Feedback = bind(owner, bindings)
        let sourceSystem = self.system
        return ObservableSystem { feedback in sourceSystem([newFeedback] + feedback) }
    }

    // ... other operator

    // There are some duplicate code in each operator, 
    // It's fine in the demo context since this will improve readabylity.
}

The ObservableSystem is like Observable in Rx.

And reacted, binded is like Operators in Rx.

Now the system can be chainable:

ObservableSystem.create(
    initialState: State.humanHasIt,
    reduce: { (state: State, event: Event) -> State in
        switch event {
        case .throwToMachine:
            return .machineHasIt
        case .throwToHuman:
            return .humanHasIt
        }
},
    scheduler: MainScheduler.instance
    )
    .binded(self) { ... }
    .reacted(request: { $0.machinePitching }, effects: { ... })
    .reacted(request: { ... }, effects: { ... })
    .reacted(request: { ... }, effects: { ... })
    .system([])
    .subscribe()
    .disposed(by: disposeBag)

It will bring us some benefits:

• system has its own namespace ObservableSystem
• more consist with Rx
• easier to add operator
• less typing

With the benefits, I proposal to add this feature.

A running example can be found here with commit: introduce ObservableSystem. It also handle driver version (DriverSystem).

I'm open to disccuss 😄, If this is accepted, I will make a PR.

Thanks.

[eliekarouz]

Hi @beeth0ven , thanks for this proposal. I was thinking how you would be able to test effects/feedbacks?

[beeth0ven]

Hi @eliekarouz, thanks for your interest.

Effects can be tested as before with TestScheduler:

1. create TestScheduler
2. create mocked effects
3. create mocked events
4. inject mocked effects and events to system
5. assert output states

PlayCatch Test

let events = [
   "tm" : Event.throwToMachine,
   "th" : .throwToHuman,
]

let states = [
   "h" : State.humanHasIt,
   "m" : .machineHasIt
]

// 1. create `TestScheduler`
let scheduler = TestScheduler(initialClock: 0, resolution: resolution, simulateProcessingDelay: false)

//  2. create mocked effects
let mockedEffects: (PitchRequest) -> Observable<Event> = scheduler.mock(values: events) { _ -> String in
   return "----th"
};


// 3. create mocked events
let (
   inputEvents,
   expectedStates
   ) = (
   scheduler.parseEventsAndTimes(timeline: "------tm------tm------tm-------", values: events).first!,
   scheduler.parseEventsAndTimes(timeline: "h-----m---h---m---h---m---h----", values: states).first!
)

// 4. inject mocked effects and events to system
let observableSystem = ObservableSystem.create(
   initialState: State.humanHasIt,
   reduce: { (state: State, event: Event) -> State in
       switch event {
       case .throwToMachine:
           return .machineHasIt
       case .throwToHuman:
           return .humanHasIt
       }
},
   scheduler: scheduler
   )
   .reacted(request: { $0.machinePitching }, effects: mockedEffects)

let state = observableSystem.system([{ _ in scheduler.createHotObservable(inputEvents).asObservable() }])

let recordedState = scheduler.record(source: state)

scheduler.start()

// 5. assert output states
XCTAssertEqual(recordedState.events, expectedStates)

This example use MarbleTests which can be found in RxExample_iOSTests.

[beeth0ven]

Hi there!

Long time no see, hoping every one is doing well. I missed all of you!

Things get evoluted after this proposal. I'm happy to see swift-composable-architecture use a similar pattern and become popular, that's pretty cool!

Then I tried to evolute this idea, and open source a library called love.dart 😄. Yeah it's written in dart since I developed flutter apps recently.

If you are still interested with this "operator pattern". Feel free to take a look. Feedback 😄 are also welcome!

Thank you!

Best Wishes!