Можно только добавить блоки уведомлений внутри runloops

Question

я несколько аварий на этой части кода:

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); 

Я думаю, что проблема в том, что я нахожусь не на MainScheduler.instance и rxRealm смотреть на main thread, но я не хочу этого. Можно ли это исправить?

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 в доставляются по умолчанию в главном потоке (где установлен runloop для приложения по умолчанию). Вот почему в вашем коде вы получаете исключение «Можете только добавлять блоки уведомлений из runloops», потому что вы пытаетесь подписаться на уведомления в фоновом потоке без runloop.

Когда вы используете Observable.from(... some realm collection ...), RxRealm подписывается на уведомления, которые отправляет сама Realm, и это когда вы получаете исключение, потому что ваш код явно переключается на фоновый поток в этой строке .observeOn(ConcurrentDispatchQueueScheduler.init(queue....

Вы можете пойти двумя разными путями, чтобы решить вашу текущую проблему:

1) Не переключаться на фоне очереди перед вашим оператором subscribe (при условии, что вы делаете подписку Rx на главном потоке)

2) Внутри вашего метода fetchTrainRx используйте DispatchQueue.main.sync, чтобы создать Наблюдаемый на основной теме.

Это должно решить вашу текущую проблему.

В любом случае - из вашего кода я вижу, что вы храните ссылку Realm где-то в своем классе (или, по крайней мере, похоже, что вы это делаете), что не является лучшей практикой. Если вы работаете над разными потоками, используйте let realm = try! Realm() для получения ссылки Realm на текущий поток каждый раз; это гарантирует, что вы всегда будете работать с правильным экземпляром области.