RxSwift subscribeOn and observeOn not on the expected background Thread

user_326036 0
Viewed 633

I have the following scheduler:

let scheduler = ConcurrentDispatchQueueScheduler(queue: .global(qos: .background))

tried with: observeOn(scheduler) also with: subscribeOn(scheduler)

I'm expecting that the callback from the subscribe would be executed on some the background thread, because of the scheduler.

func signIn(withExternalUserID userID: UserID, authenticationToken: String) -> Single<QBUUser> {
        let authenticationToken = "Bearer \(authenticationToken)"
        return .create { observable in
            let request = QBRequest.logIn(
                withUserLogin: "\(userID)",
                password: authenticationToken,
                successBlock: { _, user in
                    user.password = QBSession.current.sessionDetails?.token
                    observable(.success(user)) // <- Here we are on the main thread
                }, errorBlock: { [unowned self] in
                    observable(.error(ChatError.signInError(self.resolve(errorResponse: $0))))  // <- same here: main thread
                }
            )
            return Disposables.create {
                request.cancel()
            }
        }
    }

Because the callbacks of the internal API are executed in the main thread, all my subcribes instead of background thread are also executed on the main thread.

tokenRequest
      .flatMap { [unowned self] token -> Single<QBUUser> in
           return self.chatService.signIn(withExternalUserID: currentUser.id, authenticationToken: token)
      }
      .flatMap { [unowned self] user -> Single<QBUUser> in
           return self.chatService.connect(user: user).andThen(Single.just(user))
      }
      .observeOn(self.scheduler)
      .subscribe(onNext: { [unowned self] user in
            self.chatStorage.set(currentQBUser: user, currentUser: currentUser)
            completable(.completed)
            self.loginSempahore.signal()
       }, onError: { [unowned self] error in
            self.chatStorage.unsetCurrentQBUser()
            completable(.error(error))
            self.loginSempahore.signal()
       })
       .disposed(by: self.disposeBag)

What is the correct way to force all callbacks from flatMap, and subscribe being executed by background thread?

rx-swift
1 Answers

There is no way to force a function to call its callback on a particular thread unless the library gives you the option to do it. Your successBlock and errorBlock closures will be called on whatever thread QBRequest.logIn wants to call them on and there's nothing you can do about that.

That said, the .observeOn(_:) operator will transfer the execution to a different thread so in your last code example, your onNext and onError closures will be executed on the self.scheduler thread.

user_5064410
Related