NetworkOnMainThreadException в flatMap

Question

Я новичок в rxjava или rxandroid, и я хочу изменить AsyncTask и обратного вызова ад observable и subscriber. Но у меня проблема с этим. Мне нужно сделать два запроса в моей базе данных.

Мой ответ с первого запроса будет результатом второго запроса. Я пытаюсь разрешить это с помощью flatMap. Первый запрос возвращает значение, и все в порядке, но следующий запрос дает мне NetworkOnMainThreadException.

Я знаю, что запрос выполняется на main thread, но почему? Я пытаюсь добавить subscribeOn(Schedulers.io()) до flatMap, но результат будет таким же. Не могли бы вы помочь мне в этом и объяснить, что я делаю неправильно? Заранее спасибо. мой код ......

private void getFavouriteList(){ 
    Observable.create((Observable.OnSubscribe<PaginatedScanList<UserDO>>) subscriber -> { 
     final Map<String, AttributeValue> filterExpressionAttributeValues = new HashMap<>(); 
     filterExpressionAttributeValues 
       .put(":val1", new AttributeValue().withS(sharedPreferences.getString("socialId", ""))); 
     final DynamoDBScanExpression scanExpression = new DynamoDBScanExpression() 
       .withFilterExpression("socialId = :val1") 
       .withExpressionAttributeValues(filterExpressionAttributeValues); 
     PaginatedScanList<UserDO> result = dynamoDBMapper.scan(UserDO.class, scanExpression); 
     Log.d(TAG, "first result size " + result.size()); 
     subscriber.onNext(result); 
     subscriber.onCompleted(); 
    }) 
      .subscribeOn(Schedulers.io()) 
      .observeOn(AndroidSchedulers.mainThread()) 
      .filter(result -> { 
       if(result.isEmpty()) { 
        Toast.makeText(context, "Can not find user", Toast.LENGTH_SHORT).show(); 
        return false; 
       } 
       return true; 
      }) 
      .flatMap(user -> Observable.from(user.get(0).getFavourites())) 
      .subscribeOn(Schedulers.io()) 
      .flatMap(result -> { 
       final Map<String, AttributeValue> filterExpressionAttributeValues = new HashMap<>(); 
       filterExpressionAttributeValues 
         .put(":val1", new AttributeValue().withS(result)); 
       filterExpressionAttributeValues 
         .put(":val2", new AttributeValue().withN("1")); 
       final DynamoDBScanExpression scanExpression = new DynamoDBScanExpression() 
         .withFilterExpression("productId = :val1 and selling = :val2") 
         .withExpressionAttributeValues(filterExpressionAttributeValues); 
       PaginatedScanList<ProductDO> res = dynamoDBMapper.scan(ProductDO.class, scanExpression); 
       Log.d(TAG, "second result size " + res.size()); 
       return Observable.from(res); 
      }) 
      .subscribe(new Subscriber<ProductDO>() { 
       @Override 
       public void onCompleted() { 
        favouriteProgressBar.setVisibility(View.INVISIBLE); 
       } 

       @Override 
       public void onError(Throwable e) { 
        e.printStackTrace(); 
        favouriteProgressBar.setVisibility(View.INVISIBLE); 
       } 

       @Override 
       public void onNext(ProductDO productDO) { 
        Log.d(TAG, "productId " + productDO.getProductId()); 
        product.add(productDO); 
        adapter.notifyDataSetChanged(); 
       } 
      }); 
} 

Answer 1

Переместить ваш .observeOn(AndroidSchedulers.mainThread()) только перед вашим подписаться.

PS: Вы не должны использовать Observable.create(), если нет другой альтернативы.

Отредактировано для исправления тоста в проблеме фильтра.

private void getFavouriteList(){ 
    Observable.create((Observable.OnSubscribe<PaginatedScanList<UserDO>>) subscriber -> { 
     final Map<String, AttributeValue> filterExpressionAttributeValues = new HashMap<>(); 
     filterExpressionAttributeValues 
       .put(":val1", new AttributeValue().withS(sharedPreferences.getString("socialId", ""))); 
     final DynamoDBScanExpression scanExpression = new DynamoDBScanExpression() 
       .withFilterExpression("socialId = :val1") 
       .withExpressionAttributeValues(filterExpressionAttributeValues); 
     PaginatedScanList<UserDO> result = dynamoDBMapper.scan(UserDO.class, scanExpression); 
     Log.d(TAG, "first result size " + result.size()); 
     subscriber.onNext(result); 
     subscriber.onCompleted(); 
    }) 
      .subscribeOn(Schedulers.io()) 
      .observeOn(AndroidSchedulers.mainThread()) 
      .filter(result -> { 
       if(result.isEmpty()) { 
        Toast.makeText(context, "Can not find user", Toast.LENGTH_SHORT).show(); 
        return false; 
       } 
       return true; 
      }) 
      .observeOn(Schedulers.io()) 
      .flatMap(user -> Observable.from(user.get(0).getFavourites())) 
      .flatMap(result -> { 
       final Map<String, AttributeValue> filterExpressionAttributeValues = new HashMap<>(); 
       filterExpressionAttributeValues 
         .put(":val1", new AttributeValue().withS(result)); 
       filterExpressionAttributeValues 
         .put(":val2", new AttributeValue().withN("1")); 
       final DynamoDBScanExpression scanExpression = new DynamoDBScanExpression() 
         .withFilterExpression("productId = :val1 and selling = :val2") 
         .withExpressionAttributeValues(filterExpressionAttributeValues); 
       PaginatedScanList<ProductDO> res = dynamoDBMapper.scan(ProductDO.class, scanExpression); 
       Log.d(TAG, "second result size " + res.size()); 
       return Observable.from(res); 
      }) 
      .observeOn(AndroidSchedulers.mainThread()) 
      .subscribe(new Subscriber<ProductDO>() { 
       @Override 
       public void onCompleted() { 
        favouriteProgressBar.setVisibility(View.INVISIBLE); 
       } 

       @Override 
       public void onError(Throwable e) { 
        e.printStackTrace(); 
        favouriteProgressBar.setVisibility(View.INVISIBLE); 
       } 

       @Override 
       public void onNext(ProductDO productDO) { 
        Log.d(TAG, "productId " + productDO.getProductId()); 
        product.add(productDO); 
        adapter.notifyDataSetChanged(); 
       } 
      }); 
}