Android RxJava асинхронный вызов в функции карты

Question

Об изменении «SortBy» моя программа сделает NetworkIO, чтобы получить лучшие фильмы и отобразить их.

Тем не менее, кажется, что, хотя я сделал subscribeOn(Schedulers.io()), в NetworkIO MovieDB.getPopular() и MovieDB.getTopRated() в функции call в map являются excuted в основном потоке, и я получаю android.os.NetworkOnMainThreadException.

Мне было интересно, как сделать public Movie[] call(SortBy sortBy) асинхронным.

sortObservable.map(new Func1<SortBy, Movie[]>() { 
    @Override 
    public Movie[] call(SortBy sortBy) { 
     try { 
      switch (sortBy) { 
       case POPULAR: 
        return MovieDB.getPopular(); // NETWORK IO 
       case TOP_RATED: 
        return MovieDB.getTopRated(); // NETWORK IO 
      } 
     } catch (IOException e) { 
      e.printStackTrace(); 
     } catch (JSONException e) { 
      e.printStackTrace(); 
     } 
     return new Movie[0]; 
    } 
}) 
     .subscribeOn(Schedulers.io()) 
     .observeOn(AndroidSchedulers.mainThread()) 
     .subscribe(new Action1<Movie[]>() { 
      @Override 
      public void call(Movie[] movies) { 
       imageAdapter.loadData(movies); 
      } 
     }); 

Answer 1

Наконец я разбирайтесь самостоятельно:

sortObservable.flatMap(new Func1<SortBy, Observable<Movie[]>>() { 

    @Override 
    public Observable<Movie[]> call(SortBy sortBy) { 
     switch (sortBy) { 
      case POPULAR: 
       return Observable.fromCallable(() -> MovieDB.getPopular()).subscribeOn(Schedulers.io()); 
      case TOP_RATED: 
       return Observable.fromCallable(() -> MovieDB.getTopRated()).subscribeOn(Schedulers.io()); 
      default: 
       return Observable.fromCallable(() -> new Movie[0]).subscribeOn(Schedulers.io()); 
     } 
    } 
}) 
     .observeOn(AndroidSchedulers.mainThread()) 
     .subscribe(new Action1<Movie[]>() { 
      @Override 
      public void call(Movie[] movies) { 
       imageAdapter.loadData(movies); 
      } 
     }); 

Answer 2

Проверьте, подходит ли нижеследующее для вас. Он использует flatMap вместо map.

sortObservable.flatMap(new Func1<SortBy, Observable<Movie[]>>() { 

     @Override 
     public Observable<Movie[]> call(SortBy sortBy) { 
      try { 
       switch (sortBy) { 
        case POPULAR: 
         return Observable.just(MovieDB.getPopular()); // NETWORK IO 
        case TOP_RATED: 
         return Observable.just(MovieDB.getTopRated()); // NETWORK IO 
       } 
      } catch (IOException e) { 
       e.printStackTrace(); 
      } catch (JSONException e) { 
       e.printStackTrace(); 
      } 
      return Observable.just(new Movie[0]); 
     } 
    }).subscribe(new Action1<Movie[]>() { 
     @Override 
     public void call(Movie[] movies) { 
      imageAdapter.loadData(movies); 
     } 
    }); 

Из исходного кода на Github, кажется, что вы используете синхронный режим выполнения запросов с использованием OkHttp. OkHttp также поддерживает асинхронные запросы, и это может быть предпочтительнее. Ниже будут изменения, требуемые в некоторых из методов.

1. run метод должен потреблять enqueue вместо execute.

   Observable<String> runAsync(String url){ 
   return Observable.create(subscriber -> { 
       Request request = new Request.Builder().url(url).build(); 
   
       client.newCall(request).enqueue(new Callback() { 
   
        @Override 
        public void onResponse(Call call, Response response) throws IOException { 
         subscriber.onNext(response.body().string()); 
        } 
   
        @Override 
        public void onFailure(Call call, IOException e) { 
         subscriber.onError(e); 
        } 
       }); 
   }); 
   } 
   

2. getApi может возвращать Observable<Movie[]> вместо Movie[]

   public Observable<Movie[]> getApiAsync(String type){ 
   return runAsync("http://api.themoviedb.org/3/movie/" + type 
        + "?api_key=412e9780d02673b7599233b1636a0f0e").flatMap(response -> { 
         Gson gson = new Gson(); 
         Map<String, Object> map = gson.fromJson(response, 
           new TypeToken<Map<String, Object>>() { 
           }.getType()); 
         Movie[] movies = gson.fromJson(gson.toJson(map.get("results")), 
           Movie[].class); 
         return Observable.just(movies); 
        }); 
   }