private Observable < Boolean > synchroniseEvents ( final RxComapiClient client , @ NonNull final List < ChatConversation > conversationsToUpdate , @ NonNull final List < Boolean > successes ) { final List < ChatConversation > limited = limitNumberOfConversations ( conversationsToUpdate ) ; if ( limited . isEmpty ( ) ) { return Observable . fromCallable ( ( ) -> true ) ; } return Observable . from ( limited ) . onBackpressureBuffer ( ) . flatMap ( conversation -> persistenceController . getConversation ( conversation . getConversationId ( ) ) ) . flatMap ( conversation -> { if ( conversation . getLastRemoteEventId ( ) > conversation . getLastLocalEventId ( ) ) { final long from = conversation . getLastLocalEventId ( ) >= 0 ? conversation . getLastLocalEventId ( ) : 0 ; return queryEventsRecursively ( client , conversation . getConversationId ( ) , from , 0 , successes ) . map ( ComapiResult :: isSuccessful ) ; } else { return Observable . fromCallable ( ( Callable < Object > ) ( ) -> true ) ; } } ) . flatMap ( res -> Observable . from ( successes ) . all ( Boolean :: booleanValue ) ) ; }
Synchronise missing events for the list of locally stored conversations .