Hot questions for Using RxJava 2 in arraylist

Question:

WAnt to filter list with RxJava2. somehow got it working. but the prob is, it is returning only one item in List in Consumer callback(size is always 1)

here is the code:

 Observable.fromIterable(arraylist)
                   .filter(new Predicate<BasicListModel>() {
                       @Override
                       public boolean test(BasicListModel model) throws Exception {
                           return true; //returning true for all items in filter
                       }
                   })
                   .toList()
                   .observeOn(Schedulers.computation())
                   .subscribe(new Consumer<List<BasicListModel>>() {
                       @Override
                       public void accept(List<BasicListModel> models) throws Exception {

                               Log.i("TAG"," size:"+models.size());

                       }
                   });

i am new to RxJAva1 or RxJava2.


Answer:

You can filter with proper iteration as followed

Observable.fromIterable(offerBasicListModel)
            .observeOn(Schedulers.computation())
            .filter(new Predicate<BasicListModel>() {
                @Override
                public boolean test(BasicListModel model) throws Exception {

                    if (model.isDownloading()) //assume
                        return true; // if true, object will redirect to `doOnNext`
                    else
                        return false;
                }
            })
            .doOnNext(new Consumer<BasicListModel>() {
                @Override
                public void accept(BasicListModel model) throws Exception {
                    Log.d("objects one by one ->",model.getId());
                }
            })
            .toList()
            .subscribe(new Consumer<List<BasicListModel>>() {
                @Override
                public void accept(List<BasicListModel> model) throws Exception {
                    Log.d(TAG, "filtered list size: "+model.size());
                }
            });

if you're supporting java 8, then

Observable.fromIterable(offerBasicListModel)
            .observeOn(Schedulers.computation())
            .filter(BasicListModel::isDownloading)
            .doOnNext(
            model ->Log.d(TAG,"filtered objects one by one ->",model.getId())
            )
            .toList()
            .subscribe(model -> Log.d(TAG, "filtered list size: "+model.size()));

UPDATE

Custom User model

public class UserModel {

private int age;
private String name;

public UserModel(int age, String name) {
    this.age = age;
    this.name = name;
}

public int getAge() {
    return age;
}

public void setAge(int age) {
    this.age = age;
}

public String getName() {
    return name;
}

public void setName(String name) {
    this.name = name;
} 
}

perform filtering user with age above 18

ArrayList<UserModel> userModelArrayList = new ArrayList<>();
    UserModel user;
    for (int i = 0; i < 20; i++) {
        if (i % 2 == 0)
            user = new UserModel(25, "user" + i);
        else
            user = new UserModel(15, "user" + i);

        userModelArrayList.add(user);
    }

    io.reactivex.Observable
            .fromIterable(userModelArrayList)
            .observeOn(AndroidSchedulers.mainThread())
            .subscribeOn(Schedulers.computation())
            .filter(userModel -> userModel.getAge() > 18)
            .toList()
            .subscribe(new SingleObserver<List<UserModel>>() {
                @Override
                public void onSubscribe(Disposable d) {
                    /* to do */
                }

                @Override
                public void onSuccess(List<UserModel> userModels) {
                    Log.d("userModels", " after filtering: " + userModels.size());

                }

                @Override
                public void onError(Throwable e) {
                    /* to do */
                }
            });

Here I am able to achieve 10 filtered user objects in onSuccess().

I suggest you to try this approach first using sample code if it is working then you can modify and trace where exactly you're doing wrong.

Question:

I have a class Person which has isChild attribute along with other attributes

class Person {
  boolean isChild;
  //Other attributes.
}

Now I have a method with takes a list of Person as input. My output should also return a list, but the list should have all the Person objects first for whom isChild is true, followed by other Person objects. In java, I can do it in the following way.

public List<Person> returnListWithChildFirst(@NonNull List<Person> personList) {

    List<Person> childList = new ArrayList<>();
    List<Person> nonChildList = new ArrayList<>();

    for (Person person : personList) {
      if (person.isChild) {
        childList.add(person);
      } else {
        nonChildList.add(person);
      }
    }

    childList.addAll(nonChildList);
    return childList;
  }

What is the best way to do this in RxJava2?


Answer:

You can do it like this;

Observable.fromIterable(personList).sorted( (p1, p2) -> Boolean.compare(!p1.isChild(),!p2.isChild())).toList().blockingGet();

Or with java 8

personList.stream().sorted(Comparator.comparing(Person::isChild).reversed()).collect(Collectors.toList());

Question:

I have an arraylist which contains a set of objects.I would like to use rxjava so i can loop through the list in such a way that in the onSubscribe method instead of getting the entire list at once i get each list item 1 at a time


Answer:

It was actually Observable.fromIterable that did the trick, Observable.from returns the entire array at a time.

Question:

I'm still learning RxJava, I know the basic what is Observable & Observers, But still confused how, where & when to use filter, map, flatmap, etc

Do you have any good suggest how to convert this code to RxJava ?

void locationChanged(LocationResult locationResult) {
        for (Location location : locationResult.getLocations()) {
        Double lat = location.getLatitude();
        Double lng = location.getLongitude();

        boolean isMock;
        if (Build.VERSION.SDK_INT >= 18) {
            isMock = location.isFromMockProvider();
        } else {
            isMock = !Settings.Secure.getString(getContentResolver(), Settings.Secure.ALLOW_MOCK_LOCATION).equals("0");
        }

        for (int k = 0; k < offices.size(); k++) {
            Office office = offices.get(k);

            Double d_lat = Double.valueOf(office.getOc_lat());
            Double d_long = Double.valueOf(office.getOc_long());
            Double d_radius = Double.valueOf(office.getOc_radius());

            Location.distanceBetween(lat, lng, d_lat, d_long, resultApi);
            String s_distanceToOffice = String.valueOf(resultApi[0]);
            Double d_distanceToOffice = Double.parseDouble(s_distanceToOffice);
            //Log.e(TAG, "locationChanged: Distance -> " + d_distanceToOffice);

            // check-in in radius
            if (d_distanceToOffice < d_radius) {
                buttonOutOfRadius.setVisibility(View.GONE);
                progressBarPosition.setVisibility(View.GONE);
                buttonRecord.setVisibility(View.VISIBLE);

                // checking fake gps
                if (isMock) {
                    processTheMock();
                } else {
                    in_area = "Y";

                    site_name = office.getOc_site();
                    site_id = office.getOc_id();
                    site_lat = office.getOc_lat();
                    site_long = office.getOc_long();

                    processNoMock(lat, lng, site_name);
                }
                stopLocationUpdates();
                break;
            }

            else if (d_distanceToOffice > d_radius) {
                progressBarPosition.setVisibility(View.GONE);
                buttonOutOfRadius.setVisibility(View.VISIBLE);
                buttonRecord.setVisibility(View.GONE);
            }
        }

    }
    stopLocationUpdates();
}

Which code needs to be changed and unnecessary to change ? What I've done like this :

Observable.fromIterable(offices)
                    .subscribeOn(Schedulers.io())
                    .observeOn(AndroidSchedulers.mainThread())
                    .subscribe()

do i need to just copy paste the code inside looping "for(..) into "oNext(Office office)" ?

It's nice to know from your experience.


Answer:

After few days i got the the good flow to refactor it into Rx,

Here my code in kotlin, easy to read, if you have any better alternative for this, please share to me

private fun locationChanged(locationResult: LocationResult) {
    for (location in locationResult.locations) {

        val lat = location.latitude
        val lng = location.longitude
        var dDistanceToOffice: Double
        var dRadius: Double

        Observable.fromIterable(offices)
                .subscribeOn(Schedulers.newThread())
                .filter {
                    val dLat = it.oc_lat!!.toDouble()
                    val dLong = it.oc_long!!.toDouble()
                    dRadius = it.oc_radius!!.toDouble()

                    // Check the distance "MyLocation" to "OfficeLocation"
                    Location.distanceBetween(lat, lng, dLat, dLong, resultApi)
                    val sDistanceToOffice = resultApi[0].toString()
                    dDistanceToOffice = sDistanceToOffice.toDouble()

                    dDistanceToOffice < dRadius
                }
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe(object : Observer<Office> {
                    override fun onSubscribe(d: Disposable) {
                        disposable = d
                    }

                    override fun onNext(office: Office) {
                        inArea = "Y"
                        siteName = office.oc_site
                        siteId = office.oc_id
                        siteLat = office.oc_lat!!
                        siteLong = office.oc_long!!
                    }

                    override fun onError(e: Throwable) {
                        Log.e(TAG, "locationChanged() Error: ${e.message}")
                    }

                    override fun onComplete() {
                        // checking the data in Area = N or Y
                        if (isMock(location)) {
                            processTheMock()
                        } else {
                            if (inArea == "Y") {
                                hideButtonOutRadius()
                                hideProgressbarPosition()
                                showButtonCheckin()
                                processNoMock(lat, lng, "In Radius", siteName!!)
                            } else {
                                retrofitGoogleGeocoding(lat, lng)
                                inArea = "N"
                                siteName = newAddress
                                siteLat = lat.toString()
                                siteLong = lng.toString()
                                hideButtonOutRadius()
                                hideProgressbarPosition()
                                showButtonCheckin()
                                processNoMock(lat, lng, "In Radius", siteName!!)
                            }
                        }
                        stopLocationUpdates()
                    }
                })
    } // end of location
}