android - OnErrorNotImplementedException Caused by: rx.exceptions.MissingBackpressureException -

i have error when call syncusers() method.

rx.exceptions.onerrornotimplementedexception @ rx.observable$27.onerror(observable.java:8139)     @ rx.observers.safesubscriber._onerror(safesubscriber.java:157)     @ rx.observers.safesubscriber.onerror(safesubscriber.java:120)     @ rx.internal.operators.operatorzip$zip$innersubscriber.onerror(operatorzip.java:325)     @ rx.internal.operators.operatorzip$zip$innersubscriber.onnext(operatorzip.java:333)     @ io.realm.rx.realmobservablefactory$5$1.onchange(realmobservablefactory.java:143)     @ io.realm.rx.realmobservablefactory$5$1.onchange(realmobservablefactory.java:139)     @ io.realm.realmresults.notifychangelisteners(realmresults.java:1010)     @ io.realm.realmresults.notifychangelisteners(realmresults.java:996)     @ io.realm.handlercontroller.notifyrealmresultscallbacks(handlercontroller.java:303)     @ io.realm.handlercontroller.notifysyncrealmresultscallbacks(handlercontroller.java:284)     @ io.realm.handlercontroller.notifytypebasedlisteners(handlercontroller.java:275)     @ io.realm.handlercontroller.notifyalllisteners(handlercontroller.java:262)     @ io.realm.handlercontroller.realmchanged(handlercontroller.java:385)     @ io.realm.handlercontroller.handlemessage(handlercontroller.java:116)     @ android.os.handler.dispatchmessage(handler.java:98)     @ android.os.looper.loop(looper.java:148)     @ android.app.activitythread.main(activitythread.java:5417)     @ java.lang.reflect.method.invoke(native method)     @ com.android.internal.os.zygoteinit$methodandargscaller.run(zygoteinit.java:726)     @ com.android.internal.os.zygoteinit.main(zygoteinit.java:616) caused by: rx.exceptions.missingbackpressureexception     @ rx.internal.util.rxringbuffer.onnext(rxringbuffer.java:352)     @ rx.internal.operators.operatorzip$zip$innersubscriber.onnext(operatorzip.java:331)     @ io.realm.rx.realmobservablefactory$5$1.onchange(realmobservablefactory.java:143)      @ io.realm.rx.realmobservablefactory$5$1.onchange(realmobservablefactory.java:139)      @ io.realm.realmresults.notifychangelisteners(realmresults.java:1010)      @ io.realm.realmresults.notifychangelisteners(realmresults.java:996)      @ io.realm.handlercontroller.notifyrealmresultscallbacks(handlercontroller.java:303)      @ io.realm.handlercontroller.notifysyncrealmresultscallbacks(handlercontroller.java:284)      @ io.realm.handlercontroller.notifytypebasedlisteners(handlercontroller.java:275)      @ io.realm.handlercontroller.notifyalllisteners(handlercontroller.java:262)      @ io.realm.handlercontroller.realmchanged(handlercontroller.java:385)      @ io.realm.handlercontroller.handlemessage(handlercontroller.java:116)      @ android.os.handler.dispatchmessage(handler.java:98)      @ android.os.looper.loop(looper.java:148)      @ android.app.activitythread.main(activitythread.java:5417)      @ java.lang.reflect.method.invoke(native method)      @ com.android.internal.os.zygoteinit$methodandargscaller.run(zygoteinit.java:726)      @ com.android.internal.os.zygoteinit.main(zygoteinit.java:616) 

the problem seems come first operation.

public observable<void> syncusers() {             return client.getinstance().getusers()                 .flatmap(new func1<users, observable<users>>() {                     @override                     public observable<users> call(final users users) {                          users.setidentifier(prefhelper.getusername()); //primary key                         final realm realm = realm.getdefaultinstance();                         realm.executetransaction(new realm.transaction() {                             @override                             public void execute(realm realm) {                                 realm.copytorealmorupdate(users);                             }                         });                         realm.close();                         return observable.just(users);                     }                 })                 .flatmap(new func1<users, observable<void>>() {                     @override                     public observable<void> call(final users users) {                         if (users != null && users.getusers() != null) {                             (final user user : users.getusers()) {                                 realm realm = realm.getdefaultinstance();                                 realm.executetransaction(new realm.transaction() {                                     @override                                     public void execute(realm realm) {                                         userfactory.attachparentuser(user);                                         realm.copytorealmorupdate(user);                                     }                                 });                                 realm.close();                             }                         }                         return observable.empty();                     }                 });  }      

and call :

 mgetuserssubscription = new getusersrequest(getapplicationcontext(),    mcurrentuser != null ? mcurrentuser.getidentifier() : museridtosearch, false).execute()             .subscribeon(schedulers.io())             .observeon(androidschedulers.mainthread())             .subscribe(new subscriber<void>() {                 @override                 public void oncompleted() {                     //do nothing                 }                  @override                 public void onerror(throwable e) {                     mfilesemptyview.showerror();                     mswiperefreshlayout.setrefreshing(false);                 }                  @override                 public void onnext(void avoid) {                     mswiperefreshlayout.setrefreshing(false);                     refreshdatafromdb();                 }             }); 

1.) open transaction , iterate , commit.

don't iterate, open transaction , commit.

2.) apply onbackpressurelatest() operator