Networkonmainthreadexception in Android flatmap

I am new to rxjava or rxandroid. I want to change asynctask and callback to observable and subscriber. But I have questions about this. I need to perform two requests on the database

My reply to the first request will be the result of the second request. I try to use flatmap to solve this problem. The first request returns the value, everything is normal, but the next request gives me networkonmainthreadexception

I know the request is executed on the main thread, but why? I tried to add subscribeon (schedulers. Io()) before flatmap, but the results were the same. Can you help me and explain what I did wrong? Thanks in advance. My code

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 one rror(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();
                }
            });
}

resolvent:

Move your. Observeon (androidschedulers. Mainthread()) before subscription

PS: observable. Create() should not be used unless there is no other choice

Edit to resolve toast in filter issues

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 one rror(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();
                }
            });
}