RxJava runs on the main thread instead of background thread

Question

I am using RxJava2 to fetch all the contacts and display it in a TextView. I am new to RxJava2 and due to lack of documentation samples, finding it difficult.

Here is my code to fetch contacts list from user phone using RxJava2

public class MainActivity extends AppCompatActivity {

    private TextView textView;
    private Disposable disposable;
    private List<ContactsPOJO> contactsPOJOList;
    private Observable<List<ContactsPOJO>> contactObservable;
    private Observer<List<ContactsPOJO>> contactObserver;

    @SuppressLint("CheckResult")
    @Override
    protected void onCreate(Bundle savedInstanceState) {
        super.onCreate(savedInstanceState);
        setContentView(R.layout.activity_main);
        textView = findViewById(R.id.tv);
        contactsPOJOList = new ArrayList<>();

        if (Build.VERSION.SDK_INT >= Build.VERSION_CODES.M) {
            if (checkSelfPermission(Manifest.permission.READ_CONTACTS) != PackageManager.PERMISSION_GRANTED) {
                requestPermissions(new String[]{Manifest.permission.READ_CONTACTS}, 100);
            } else {
                contactObservable = getContactObervable();

                contactObserver = getContactObserver();
                contactObservable.subscribeOn(Schedulers.io()).observeOn(AndroidSchedulers.mainThread())
                        .filter(contactsPOJOS -> {
                            if(contactsPOJOS.size()>0){
                                Collections.sort(contactsPOJOS, (o1, o2) -> o1.getName().compareTo(o2.getName()));
                            }
                            //Java 8 and API 24
                            //contactsPOJOS.stream().sorted((object1,object2) -> object1.getName().compareTo(object2.getName()));
                            return true;
                        })
                        .subscribeWith(contactObserver);

                Log.i("TAG", "onCreate1: 1");


            }
        } else {
            contactObservable = getContactObervable();
            contactObserver = getContactObserver();
            contactObservable.subscribeOn(Schedulers.io()).observeOn(AndroidSchedulers.mainThread()).subscribeWith(contactObserver);
            Log.i("TAG", "onCreate1: 2");
             }
    }

    private Observer<List<ContactsPOJO>> getContactObserver() {
        Log.i("MainActivity1",""+Thread.currentThread().getName());
        return new Observer<List<ContactsPOJO>>() {
            @Override
            public void onSubscribe(Disposable d) {
                disposable = d;
            }

            @Override
            public void onNext(List<ContactsPOJO> contactsPOJOS) {
                for(ContactsPOJO i:contactsPOJOS){
                    textView.append(i.getName()+"\n"+i.getContactNo()+"\n");
                    Log.i("TAG", "onNext: "+i.getName()+"\n"+i.getContactNo()+"\n");
                }
            }

            @Override
            public void onError(Throwable e) {

            }

            @Override
            public void onComplete() {
                Log.i("TAG", "onComplete:All items emitted ");
            }
        };

    }

    private Observable<List<ContactsPOJO>> getContactObervable() {
        Log.i("MainActivity3",""+Thread.currentThread().getName());
        return Observable.just(getContact());
    }

    public List<ContactsPOJO> getContact() {
        Log.i("MainActivity3",""+Thread.currentThread().getName());
        ContentResolver cr = getContentResolver();
        Cursor cur = cr.query(ContactsContract.Contacts.CONTENT_URI, null, null, null, null);
        if (cur != null && cur.getCount() > 0) {
            while (cur.moveToNext()) {
                String id = cur.getString(cur.getColumnIndex(ContactsContract.Contacts._ID));
                String name = cur.getString(cur.getColumnIndex(ContactsContract.Contacts.DISPLAY_NAME));

                if (Integer.parseInt(cur.getString(cur.getColumnIndex(ContactsContract.Contacts.HAS_PHONE_NUMBER))) > 0) {
                    Cursor pCur = cr.query(ContactsContract.CommonDataKinds.Phone.CONTENT_URI,
                            null,
                            ContactsContract.CommonDataKinds.Phone.CONTACT_ID + " = ?",
                            new String[]{id},
                            null);
                    while (Objects.requireNonNull(pCur).moveToNext()) {
                        String phoneNumber = pCur.getString(pCur.getColumnIndex(ContactsContract.CommonDataKinds.Phone.NUMBER));
                        contactsPOJOList.add(new ContactsPOJO(name, phoneNumber));
                    }
                    pCur.close();
                }
            }
        }
        cur.close();
        return contactsPOJOList;
    }


    @SuppressLint("CheckResult")
    @Override
    public void onRequestPermissionsResult(int requestCode, @NonNull String[] permissions, @NonNull int[] grantResults) {
        super.onRequestPermissionsResult(requestCode, permissions, grantResults);
        if (requestCode == 100) {
            if (grantResults.length > 0 && grantResults[0] == PackageManager.PERMISSION_GRANTED) {
                contactObservable = getContactObervable();

                contactObserver = getContactObserver();
                contactObservable.subscribeOn(Schedulers.io()).observeOn(AndroidSchedulers.mainThread()).subscribeWith(contactObserver);
                Log.i("TAG", "onCreate1: 3");
            }
        }
    }

    @Override
    protected void onDestroy() {
        super.onDestroy();
        disposable.dispose();
    }
}

I tried print log statements to understand which thread I am running on and it shows I am running on the main thread always. Can somebody point me what am I doing wrong in here?

Answer 1

for RxJava2: Observable.fromIterable

Answer 2

You use Observable.just() - it means that getContact() method will be invoked before as it needs to create Observable. So your code will always be invoked on current thread (in your case it's MainThread). To solve that problem use Observable.fromCallable() instead of Observable.just()