title: OnErrorNotImplementedException using RxJava2 and Retrofit2 Mosby MVI
body: <p>I'm getting a OnErrorNotImplementedException thrown and the app crashes, despite handling the error downstream(?).</p>
  
  <p>Exception</p>
  
  <pre><code>E/AndroidRuntime: FATAL EXCEPTION: RxCachedThreadScheduler-1
  Process: pl.netlandgroup.smartsab, PID: 9920
  io.reactivex.exceptions.OnErrorNotImplementedException: HTTP 401 Unauthorized
      at io.reactivex.internal.functions.Functions$OnErrorMissingConsumer.accept(Functions.java:704)
      at io.reactivex.internal.functions.Functions$OnErrorMissingConsumer.accept(Functions.java:701)
      at io.reactivex.internal.observers.LambdaObserver.onError(LambdaObserver.java:74)
      at io.reactivex.internal.operators.observable.ObservableDoOnEach$DoOnEachObserver.onError(ObservableDoOnEach.java:119)
      at io.reactivex.internal.operators.observable.ObservableDoOnEach$DoOnEachObserver.onError(ObservableDoOnEach.java:119)
      at io.reactivex.internal.operators.observable.ObservableSubscribeOn$SubscribeOnObserver.onError(ObservableSubscribeOn.java:63)
      at retrofit2.adapter.rxjava2.BodyObservable$BodyObserver.onNext(BodyObservable.java:56)
      at retrofit2.adapter.rxjava2.BodyObservable$BodyObserver.onNext(BodyObservable.java:37)
      at retrofit2.adapter.rxjava2.CallExecuteObservable.subscribeActual(CallExecuteObservable.java:43)
      at io.reactivex.Observable.subscribe(Observable.java:10838)
      at retrofit2.adapter.rxjava2.BodyObservable.subscribeActual(BodyObservable.java:34)
      at io.reactivex.Observable.subscribe(Observable.java:10838)
      at io.reactivex.internal.operators.observable.ObservableSubscribeOn$SubscribeTask.run(ObservableSubscribeOn.java:96)
      at io.reactivex.Scheduler$DisposeTask.run(Scheduler.java:452)
      at io.reactivex.internal.schedulers.ScheduledRunnable.run(ScheduledRunnable.java:61)
      at io.reactivex.internal.schedulers.ScheduledRunnable.call(ScheduledRunnable.java:52)
      at java.util.concurrent.FutureTask.run(FutureTask.java:237)
      at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:272)
      at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1133)
      at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:607)
      at java.lang.Thread.run(Thread.java:761)
   Caused by: retrofit2.adapter.rxjava2.HttpException: HTTP 401 Unauthorized
      at retrofit2.adapter.rxjava2.BodyObservable$BodyObserver.onNext(BodyObservable.java:54)
      at retrofit2.adapter.rxjava2.BodyObservable$BodyObserver.onNext(BodyObservable.java:37) 
      at retrofit2.adapter.rxjava2.CallExecuteObservable.subscribeActual(CallExecuteObservable.java:43) 
      at io.reactivex.Observable.subscribe(Observable.java:10838) 
      at retrofit2.adapter.rxjava2.BodyObservable.subscribeActual(BodyObservable.java:34) 
      at io.reactivex.Observable.subscribe(Observable.java:10838) 
      at io.reactivex.internal.operators.observable.ObservableSubscribeOn$SubscribeTask.run(ObservableSubscribeOn.java:96) 
      at io.reactivex.Scheduler$DisposeTask.run(Scheduler.java:452) 
      at io.reactivex.internal.schedulers.ScheduledRunnable.run(ScheduledRunnable.java:61) 
      at io.reactivex.internal.schedulers.ScheduledRunnable.call(ScheduledRunnable.java:52) 
      at java.util.concurrent.FutureTask.run(FutureTask.java:237) 
      at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:272) 
      at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1133) 
      at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:607) 
      at java.lang.Thread.run(Thread.java:761) 
  </code></pre>
  
  <p>Retrofit Repository:</p>
  
  <pre><code>class RetrofitRepository (retrofit: Retrofit) {
  
      val apiService: ApiService = retrofit.create(ApiService::class.java)
      var size: Int = 0
  
      fun getMapResponse(pageIndex: Int = 0): Observable&lt;MapResponse&gt; {
          return apiService.getMapResponse(pageIndex = pageIndex)
                  .doOnError { Log.d("error", it.message) }
                  .doOnNext { Log.d("currThread", Thread.currentThread().name) }
  
      }
  
      fun getItemsFormResponses(): Observable&lt;List&lt;Item&gt;&gt; {
          val list = mutableListOf&lt;Observable&lt;List&lt;Item&gt;&gt;&gt;()
          val resp0 = getMapResponse()
          resp0.subscribe { size = it.totalCount }
          var accum = 0
          do {
              list.add(getMapResponse(accum).map { it.items })
              accum++
          } while (list.size*200 &lt; size)
          return Observable.merge(list)
      }
  }
  </code></pre>
  
  <p>This results are observed by Interactor:</p>
  
  <pre><code>class MapInteractor @Inject constructor(private val repository: RetrofitRepository) {
  
      fun getMapItems(): Observable&lt;MapViewState&gt; {
          return repository.getItemsFormResponses()
                  .map {
                      if(it.isEmpty()) {
                          return@map MapViewState.EmptyResult()
                      } else {
                          val mapItems = it.map { it.toMapItem() }
                          return@map MapViewState.MapResult(mapItems)
                      }
                  }
                  .doOnNext { Log.d("currThread", Thread.currentThread().name) }
                  .startWith(MapViewState.Loading())
                  .onErrorReturn { MapViewState.Error(it) }
      }
  }
  </code></pre>
  
  <p>The <code>onErrorReturn { MapViewState.Error(it) }</code> emits correctly (right before the app crash I can see the correct thing rendered on screen). How can I avoid this exception while still maintaining the MVI architecture?</p>
  
  <p><strong>EDIT</strong></p>
  
  <p>The <a href="https://stackoverflow.com/questions/44921529/onerrornotimplementedexception-using-rxjava2-and-retrofit2-mosby-mvi/44931809#44931809">answer provided by dimsuz</a> was the correct solution, although to achieve the merging and returning one Observable with all items it had to be modified to this:</p>
  
  <pre><code>fun getMapItems(): Observable&lt;List&lt;Item&gt;&gt; {
      return getMapResponse().map {
          val size = it.totalCount
          val list = mutableListOf&lt;Observable&lt;List&lt;Item&gt;&gt;&gt;()
          var accum = 0
          do {
              list.add(getMapResponse(accum++).map { it.items })
          } while (list.size*200 &lt; size)
          return@map list.zip { it.flatten() }
      }.mergeAll()
  }
  </code></pre>