RxSwift: implementing concatMap generating Reentrancy anomaly was detected error

Question

I'm trying to implement concatMap within RxSwift but when I tried to set a new value to the observable within the concatMap I'm getting this error:

Reentrancy anomaly was detected.
  > Debugging: To debug this issue you can set a breakpoint in /Users/SuperUser/repos/RxObserver/Pods/RxSwift/RxSwift/Rx.swift:96 and observe the call stack.
  > Problem: This behavior is breaking the observable sequence grammar. `next (error | completed)?`
    This behavior breaks the grammar because there is overlapping between sequence events.
    Observable sequence is trying to send an event before sending of previous event has finished.
  > Interpretation: This could mean that there is some kind of unexpected cyclic dependency in your code,
    or that the system is not behaving in the expected way.
  > Remedy: If this is the expected behavior this message can be suppressed by adding `.observeOn(MainScheduler.asyncInstance)`
    or by enqueuing sequence events in some other way.

Here is my code:

public let myNumbers = BehaviorRelay<String>(value: "")

override func viewDidLoad() {
    super.viewDidLoad()
    processNumbers()
    myNumbers.accept("one")
    myNumbers.accept("two")
}

func processNumbers() {
    Observable.of(myNumbers).concatMap{ $0
    }.subscribe(onNext:{
        print($0)
        if $0 == "one" || $0 == "two"{
            self.myNumbers.accept("YEAH!")
        }
        }).disposed(by: disposeBag)

}

Any of you know why I am getting this error or how can I change my implementation to avoid getting this error/warning?

I'll really appreciate your help.

Answer 1

Any of you knows why I may getting this error?

  > Problem: This behavior is breaking the observable sequence grammar. `next (error | completed)?`
    This behavior breaks the grammar because there is overlapping between sequence events.
    Observable sequence is trying to send an event before sending of previous event has finished.

how can change my implementation to avoid getting this error/warning ?

  > Remedy: If this is the expected behavior this message can be suppressed by adding `.observeOn(MainScheduler.asyncInstance)`
    or by enqueuing sequence events in some other way.

Answer 2

I found a way to fix this issue:

func processNumbers() {
        Observable.of(myNumbers).concatMap{ $0
        }
        .observeOn(MainScheduler.asyncInstance)
        .subscribe(onNext:{
            print($0)
            if $0 == "one" || $0 == "two"{
                self.myNumbers.accept("YEAH!")
            }
            }).disposed(by: disposeBag)

    }

By adding this line of code .observeOn(MainScheduler.asyncInstance) the error/warning goes away!!