RXJava как сделать наблюдаемым продолжить OnError

Question

Рассмотрим следующий пример:

WidgetObservable.text(searchView).debounce(250, TimeUnit.MILLISECONDS).flatMap(new Func1<OnTextChangeEvent, Observable<List<String>>>() { 

     @Override 
     public Observable<List<String>> call(OnTextChangeEvent onTextChangeEvent) { 
      String s = onTextChangeEvent.text().toString(); 
      return provider.getGeocodeObservable(s, 5).flatMap(new Func1<List<Address>, Observable<String>>() { 
       @Override 
       public Observable<String> call(List<Address> addresses) { 
        return Observable.from(addresses).map(new Func1<Address, String>() { 
         @Override 
         public String call(Address address) { 
          String addresss = address.getAddressLine(0); // If any additional address line present than only, check with max available address lines by getMaxAddressLineIndex() 
          String city = address.getLocality(); 
          //String state = address.getAdminArea(); 
          String country = address.getCountryName(); 
          return String.format("%s, %s, %s", addresss, city, country); 
         } 
        }); 
       } 
      }).collect(new Func0<List<String>>() { 
       @Override 
       public List<String> call() { 
        return new ArrayList<String>(); 
       } 
      }, new Action2<List<String>, String>() { 
       @Override 
       public void call(List<String> strings, String s) { 
        strings.add(s); 
       } 
      }).subscribeOn(Schedulers.io()); 
     } 
    }).observeOn(AndroidSchedulers.mainThread()).subscribe(new Subscriber<List<String>>() { 
     @Override 
     public void onCompleted() { 

     } 

     @Override 
     public void onError(Throwable e) { 
      Log.e("MapsActivity", "What?", e); 
     } 

     @Override 
     public void onNext(List<String> strings) { 
      Log.d("MapsActivity", "works " + strings); 
      adapter.clear(); 
      adapter.addAll(strings); 
      adapter.notifyDataSetChanged(); 
      searchView.showDropDown(); 
     } 
    }); 

Здесь я наблюдать текст изменений на автозаполнения зрения и действовать на него, чтобы получить соответствующие предложения адресов с помощью Android-ReactiveLocation. он отлично работает, пока ошибка в наблюдаемом геокоде не вызывает ошибки, вызывающей onError для вызова последнего абонента. С тех пор, как эта ошибка возникает, WidgetObservable больше не будет генерировать события.

Как исправить это, чтобы onError не заставил Observable прекратить излучать события изменения текста, чтобы весь поток продолжался?

BTW, Im новой для Reactive программирования и RXJava поэтому любые предложения по улучшению на мой код будет больше, то добро пожаловать :)

Answer 1

Вы можете попробовать OnErrorReturn оператор на Геокодировать наблюдаемом т.е.

provider.getGeocodeObservable(s, 5) 
     .onErrorReturn(new Func1<Throwable, List<String>>() { 
       @Override 
       public String call(Throwable throwable) { 
        return Arrays.asList(new String[]{}); 
       } 
      })