RxJava - repeat API call until result is complete

Question

I have an API that returns FamilyDetails. I use retrofit to fetch them, and it is modelled as follows:

Observable<FamilyDetails> getFamilyDetails(@Query int personId);

Once fetched, those details will be used to update a Person. My API delivers those FamilyDetails as soon as it has some information to show, which means that the details may be incomplete, and thus there is a complete = false flag delivered in responses when the server hasn't finished fetching all family details - in turn, my app should send another request to fetch the remaining FamilyDetails if we receive complete = false in the first response, and keep doing so for as long as the details remain incomplete.

I have "somewhat" achieved what I wanted, but not entirely with this code:

personApiService.getFamilyDetails(personId)
    // repeat call
    .flatMap(familyDetails -> personApiService.getFamilyDetails(personId))
    // until family details are complete
    .takeUntil(familyDetails -> familyDetails.isComplete())
    // update the person object with the family details
    .flatMap(familyDetails -> Observable.just(updatePerson(familyDetails))
    //subscribe!
    .subscribe(personSubscriber);

My personSubscriber returns an updated Person.

The problem I have with this implementation is that partial updates don't go through personSubscriber's onNext(), as I only get one call to onNext() with the updated Person object with its complete FamilyDetails.

I would like to model this using RxJava with the following requirements:

• if I receive incomplete details, I update the Person object those details belong to, and deliver it via onNext().
• If we receive incomplete details, we keep querying the API for complete details, and deliver updated Person objects via onNext() in the same observer.
• Once we get the last onNext() call with the updated Person, then onComplete() is called and we are done.

Thanks in advance!

Answer 1

personApiService.getFamilyDetails(personId)
// repeat call
.repeat()
// until family details are complete
.takeUntil(familyDetails -> familyDetails.isComplete())
//subscribe (do your updates here)
.subscribe(personSubscriber);