我刚开始接触rxjava
或者rxandroid
,并且想要用observable
和subscriber
来改善我的AsyncTask
和回调地狱。但是我遇到了一个问题:我需要向我的数据库发起两个请求。
第一个请求的响应将会成为第二个请求的结果。我尝试使用flatMap
来解决这个问题。第一个请求的返回值没问题,但是第二个请求却报错:NetworkOnMainThreadException
。
我知道请求是在主线程上执行的,但是为什么呢?我尝试在flatMap
之前添加subscribeOn(Schedulers.io())
,但结果还是一样。您能帮助我解决这个问题并解释我做错了什么吗?非常感谢。下面是我的代码......
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();
}
});
}