private Observable < ComapiResult < ConversationEventsResponse > > processEventsQueryResponse ( ComapiResult < ConversationEventsResponse > result , final List < Boolean > successes ) { ConversationEventsResponse response = result . getResult ( ) ; successes . add ( result . isSuccessful ( ) ) ; if ( response != null && response . getEventsInOrder ( ) . size ( ) > 0 ) { Collection < Event > events = response . getEventsInOrder ( ) ; List < Observable < Boolean > > list = new ArrayList <> ( ) ; for ( Event event : events ) { if ( event instanceof MessageSentEvent ) { MessageSentEvent messageEvent = ( MessageSentEvent ) event ; list . add ( persistenceController . updateStoreWithNewMessage ( ChatMessage . builder ( ) . populate ( messageEvent ) . build ( ) , noConversationListener ) ) ; } else if ( event instanceof MessageDeliveredEvent ) { list . add ( persistenceController . upsertMessageStatus ( ChatMessageStatus . builder ( ) . populate ( ( MessageDeliveredEvent ) event ) . build ( ) ) ) ; } else if ( event instanceof MessageReadEvent ) { list . add ( persistenceController . upsertMessageStatus ( ChatMessageStatus . builder ( ) . populate ( ( MessageReadEvent ) event ) . build ( ) ) ) ; } } return Observable . from ( list ) . flatMap ( task -> task ) . doOnNext ( successes :: add ) . toList ( ) . map ( results -> result ) ; } return Observable . fromCallable ( ( ) -> result ) ; }
Process the event query response . Calls appropraiate persistance controller methods for received events .