TimelineManager { void start() { disposable = Flowable.combineLatest(state(), currentTrack(), progress(), (state, track, time) -> new Timeline(state, time, track)) .observeOn(rx.io()) .flatMapCompletable(this::updateTimeline) .subscribeOn(rx.io()) .observeOn(rx.io()) .subscribe(() -> Timber.d("onCompleted"), Rx::onError); } TimelineManager(MusicController musicController, QueueManager queueManager, MediaService media,
                  Rx rx);   }
@Test public void sendRegularTimelineUpdates() { when(mockMedia.timeline(any(HttpUrl.class), anyLong(), anyString(), anyString(), anyString(), anyLong(), anyLong())) .thenReturn(Completable.complete()); timelineManager.start(); verify(mockMedia, times(1)) .timeline(TEST_URL, 100, "key", "ratingKey", "playing", 35000, 0); verify(mockMedia, times(1)) .timeline(TEST_URL, 100, "key", "ratingKey", "playing", 35000, 10000); verify(mockMedia, times(1)) .timeline(TEST_URL, 100, "key", "ratingKey", "playing", 35000, 20000); verify(mockMedia, times(1)) .timeline(TEST_URL, 100, "key", "ratingKey", "playing", 35000, 30000); }
@Test public void keepSendingUpdatesAfterTimelineError() { when(mockMedia.timeline(any(HttpUrl.class), anyLong(), anyString(), anyString(), anyString(), anyLong(), anyLong())) .thenReturn(Completable.complete()) .thenReturn(Completable.error(new IOException())) .thenReturn(Completable.error(new IOException())) .thenReturn(Completable.complete()); timelineManager.start(); verify(mockMedia, times(4)).timeline(any(HttpUrl.class), anyLong(), anyString(), anyString(), anyString(), anyLong(), anyLong()); }