Can only add notification blocks from within runloops

Question

i've some crashs on this part of code :

SRNetwork.provider
        .request(SRService.postData(user_id: userId))
        .mapArray(STrain.self)
        .observeOn(ConcurrentDispatchQueueScheduler.init(queue: SDispatchQueue.dataTrain.getQueue()))
        .subscribe({ (event) -> Void in
            switch event {
            case .next(let response):
                self.train.value = response
                SRealmTrain.sharedInstance.cacheTrain(response)
            case .error(let error):
                SRealmTrain.sharedInstance.fetchTrainRx(userId) //CRASH IS HERE
                    .bindTo(self.train)
                    .addDisposableTo(self.disposeBag)
                print("\(error)")
            default: break;
            }
        })
        .addDisposableTo(disposeBag);

I think the problem is that i'm not on the MainScheduler.instance and rxRealm watch on the main thread but i don't want it. Is it possible to fix it ?

fetchTrainRx :

public func fetchTrainRx(_ userId: String) -> Observable<[STrain]> {

    let predicate = NSPredicate(format: "userId == %@", userId)

    if let realm = realm {

        return Observable.from(realm
            .objects(SRTrain.self)
            .filter(predicate)
            .sorted(byProperty: "order", ascending: true))
            .map ({
                $0.map(STrain.init)
        })
    }
    return Observable.just([]);
}

Answer 1

Realm's notifications are delivered by default on the main thread (where the app's default runloop is installed). This is why in your code you get an exception "Can only add notification blocks from within runloops" because you try to subscribe for notifications in a background thread with no runloop.

When you use Observable.from( ... some realm collection ... ), RxRealm subscribes for the notifications that Realm itself sends and that's when you get the exception because your code explicitly switches to a background thread on this line .observeOn(ConcurrentDispatchQueueScheduler.init(queue....

You can go two separate ways to solve your current issue:

1) Do not switch to a background queue before your subscribe operator (assuming you're doing the Rx subscription on the main thread)

2) Inside your fetchTrainRx method use DispatchQueue.main.sync to create the Observable on the main thread.

This should solve your current issue.

In any case - from you code I see you store the Realm reference somewhere in your class (or at least it looks like you do that), which is not best practice. If you're working on different threads use let realm = try! Realm() to fetch a Realm reference on the current thread each time; this will ensure you always work with the correct realm instance.