Adapt step in the background causes `navigate` to be called on the same thread

[MrAsterisco]

Scenario

I am making use of RxFlow's adapt function to observe resources from a local database to decide which step should be emitted next. The observation needs to happen on a background thread, to allow the main thread to continue showing a loading animation, so I'm switching with the subscribeOn and observeOn operators. Eventually, the database retrieves the value and the appropriate step is emitted, without switching back to the main thread.

Code

func adapt(step: Step) -> Single<Step> {
   guard let step = step as? ExampleStep else { return .just(step) }
   return services.database.rx.observeSomeResource()
                .subscribeOn(/*<background thread>*/)
                .observeOn(/*<background thread>*/)
                .flatMapLatest { /*<analyze DB response and return step>*/ }
}

The problem

RxFlow then calls the navigate function, but this gets called on the same thread where the database I/O happened, which is not the main thread and causes UIKit to complain about view controllers being initialized on the wrong thread.

By looking at the first lines of lines of function coordinate(flow: Flow, with stepper: Stepper, allowStepWhenDismissed: Bool), the fix for this seems easy enough:

self.stepsRelay
            .do(onDispose: { [weak self] in
                self?.childFlowCoordinators.removeAll()
                self?.parentFlowCoordinator?.childFlowCoordinators.removeValue(forKey: self?.identifier ?? "")
            })
            .flatMapLatest { flow.adapt(step: $0) }
            .observeOn(MainScheduler.instance)
            .do(onNext: { [weak self] in self?.willNavigateRelay.accept((flow, $0)) })
            .map { return (flowContributors: flow.navigate(to: $0), step: $0) }
            .do(onNext: { [weak self] in self?.didNavigateRelay.accept((flow, $0.step)) })

Notice the observeOn before setting up the call to willNavigateRelay.

Is there a particular reason why this wasn't implemented? Is there a better way to handle this kind of situations?

[twittemb]

Hi @MrAsterisco

thanks for bringing that up. I think this is a regression in the last version.

The reactive stream used to be a Signal from the start (which guarantees no error and the main thread), and it is no longer the case since the last release. I will fix that.

thanks.

[MrAsterisco]

thanks for bringing that up. I think this is a regression in the last version.

@twittemb Ah! I was actually wondering why this never happened before, but I wasn't sure.

Thanks for taking care!

[twittemb]

@MrAsterisco

I've released a new version 2.12.2. Can you give it a try please ?

https://github.com/RxSwiftCommunity/RxFlow/releases/tag/2.12.2

[twittemb]

@MrAsterisco any update ?

[MrAsterisco]

@twittemb sorry, I couldn't find a free time slot to work on this today. I will tackle it tomorrow and update here.

[MrAsterisco]

@twittemb quick update, seems to be working as it used to! I will run some more tests over the rest of the week, but I would say we can close this issue for now. I'll reopen it if anything comes up.

Thank you!