-Functional playlist using full play queue buffering.

app/src/main/java/org/schabi/newpipe/player/BasePlayer.java

@@ -560,46 +560,61 @@ public abstract class BasePlayer implements Player.EventListener,
 
     @Override
     public void block() {
-        if (currentState != STATE_LOADING) return;
+        Log.d(TAG, "Blocking...");
 
-        changeState(STATE_LOADING);
+        if (currentState != STATE_PLAYING) return;
-        simpleExoPlayer.setPlayWhenReady(false);
+
+        simpleExoPlayer.stop();
         windowIndex = simpleExoPlayer.getCurrentWindowIndex();
         windowPos = Math.max(0, simpleExoPlayer.getContentPosition());
+
+        changeState(STATE_BUFFERING);
     }
 
     @Override
     public void unblock() {
-        if (currentState == STATE_PLAYING) return;
+        Log.d(TAG, "Unblocking...");
 
-        if (playbackManager.getMediaSource().getSize() > 0) {
+        if (currentState != STATE_BUFFERING) return;
-            simpleExoPlayer.seekToDefaultPosition();
+
-            //simpleExoPlayer.seekTo(windowIndex, windowPos);
+        if (windowIndex != playbackManager.getCurrentSourceIndex()) {
-            simpleExoPlayer.setPlayWhenReady(true);
+            windowIndex = playbackManager.getCurrentSourceIndex();
-            changeState(STATE_PLAYING);
+            windowPos = 0;
         }
+
+        simpleExoPlayer.prepare(playbackManager.getMediaSource());
+        simpleExoPlayer.seekTo(windowIndex, windowPos);
+        simpleExoPlayer.setPlayWhenReady(true);
+        changeState(STATE_PLAYING);
     }
 
     @Override
     public void sync(final int windowIndex, final long windowPos, final StreamInfo info) {
+        Log.d(TAG, "Syncing...");
+
         videoUrl = info.webpage_url;
         videoThumbnailUrl = info.thumbnail_url;
         videoTitle = info.title;
         channelName = info.uploader;
 
         if (simpleExoPlayer.getCurrentWindowIndex() != windowIndex) {
+            Log.e(TAG, "Rewinding to correct window");
             simpleExoPlayer.seekTo(windowIndex, windowPos);
         } else {
+            Log.d(TAG, "Correct window");
             simpleExoPlayer.seekTo(windowPos);
         }
     }
 
     @Override
     public void init() {
+        Log.d(TAG, "Initializing...");
+
         if (simpleExoPlayer.getPlaybackState() != Player.STATE_IDLE) simpleExoPlayer.stop();
         simpleExoPlayer.prepare(playbackManager.getMediaSource());
-        simpleExoPlayer.setPlayWhenReady(false);
+        simpleExoPlayer.seekToDefaultPosition();
-        changeState(STATE_BUFFERING);
+        simpleExoPlayer.setPlayWhenReady(true);
+        changeState(STATE_PLAYING);
     }
 
     @Override

app/src/main/java/org/schabi/newpipe/player/MediaSourceManager.java

@@ -27,6 +27,8 @@ import io.reactivex.functions.Consumer;
 
 class MediaSourceManager {
     private final String TAG = "MediaSourceManager@" + Integer.toHexString(hashCode());
+    // One-side rolling window size for default loading
+    // Effectively loads WINDOW_SIZE * 2 streams
     private static final int WINDOW_SIZE = 3;
 
     private final DynamicConcatenatingMediaSource sources;
@@ -42,13 +44,40 @@ class MediaSourceManager {
     private Subscription loadingReactor;
     private CompositeDisposable disposables;
 
+    private boolean isBlocked;
+
     interface PlaybackListener {
+        /*
+        * Called when the initial video has been loaded.
+        * Signals to the listener that the media source is prepared, and
+        * the player is ready to go.
+        * */
         void init();
 
+        /*
+        * Called when the stream at the current queue index is not ready yet.
+        * Signals to the listener to block the player from playing anything.
+        * */
         void block();
+
+        /*
+        * Called when the stream at the current queue index is ready.
+        * Signals to the listener to resume the player.
+        * May be called at any time, even when the player is unblocked.
+        * */
         void unblock();
 
+        /*
+        * Called when the queue index is refreshed.
+        * Signals to the listener to synchronize the player's window to the manager's
+        * window.
+        * */
         void sync(final int windowIndex, final long windowPos, final StreamInfo info);
+
+        /*
+        * Requests the listener to resolve a stream info into a media source respective
+        * of the listener's implementation (background, popup or main video player),
+        * */
         MediaSource sourceOf(final StreamInfo info);
     }
 
@@ -67,6 +96,13 @@ class MediaSourceManager {
                 .subscribe(getReactor());
     }
 
+    /*//////////////////////////////////////////////////////////////////////////
+    // Exposed Methods
+    //////////////////////////////////////////////////////////////////////////*/
+
+    /*
+    * Returns the media source index of the currently playing stream.
+    * */
     int getCurrentSourceIndex() {
         return sourceToQueueIndex.indexOf(playQueue.getIndex());
     }
@@ -76,6 +112,11 @@ class MediaSourceManager {
         return sources;
     }
 
+    /*
+    * Called when the player has seamlessly transitioned to another stream.
+    * Currently only expecting transitioning to the next stream and updates
+    * the play queue that a transition has occurred.
+    * */
     void refresh(final int newSourceIndex) {
         if (newSourceIndex == getCurrentSourceIndex()) return;
 
@@ -89,15 +130,107 @@ class MediaSourceManager {
         sync();
     }
 
-    private void select() {
+    void dispose() {
-        if (getCurrentSourceIndex() != -1) {
+        if (loadingReactor != null) loadingReactor.cancel();
-            sync();
+        if (playQueueReactor != null) playQueueReactor.cancel();
-        } else {
+        if (disposables != null) disposables.dispose();
-            playbackListener.block();
+
-            load();
+        loadingReactor = null;
+        playQueueReactor = null;
+        disposables = null;
+    }
+
+
+    /*//////////////////////////////////////////////////////////////////////////
+    // Event Reactor
+    //////////////////////////////////////////////////////////////////////////*/
+
+    private Subscriber<PlayQueueMessage> getReactor() {
+        return new Subscriber<PlayQueueMessage>() {
+            @Override
+            public void onSubscribe(@NonNull Subscription d) {
+                if (playQueueReactor != null) playQueueReactor.cancel();
+                playQueueReactor = d;
+                playQueueReactor.request(1);
+            }
+
+            @Override
+            public void onNext(@NonNull PlayQueueMessage event) {
+                // why no pattern matching in Java =(
+                switch (event.type()) {
+                    case INIT:
+                        init();
+                        break;
+                    case APPEND:
+                        load();
+                        break;
+                    case SELECT:
+                        onSelect();
+                        break;
+                    case REMOVE:
+                        final RemoveEvent removeEvent = (RemoveEvent) event;
+                        remove(removeEvent.index());
+                        break;
+                    case SWAP:
+                        final SwapEvent swapEvent = (SwapEvent) event;
+                        swap(swapEvent.getFrom(), swapEvent.getTo());
+                        break;
+                    case NEXT:
+                    default:
+                        break;
+                }
+
+                if (!isPlayQueueReady() && !isBlocked) {
+                    playbackListener.block();
+                    playQueue.fetch();
+                }
+                if (playQueueReactor != null) playQueueReactor.request(1);
+            }
+
+            @Override
+            public void onError(@NonNull Throwable e) {}
+
+            @Override
+            public void onComplete() {
+                dispose();
+            }
+        };
+    }
+
+    /*//////////////////////////////////////////////////////////////////////////
+    // Internal Helpers
+    //////////////////////////////////////////////////////////////////////////*/
+
+    private boolean isPlayQueueReady() {
+        return playQueue.isComplete() || playQueue.size() - playQueue.getIndex() > WINDOW_SIZE;
+    }
+
+    private boolean isCurrentIndexLoaded() {
+        return getCurrentSourceIndex() != -1;
+    }
+
+    private void tryUnblock() {
+        if (isPlayQueueReady() && isCurrentIndexLoaded() && isBlocked) {
+            isBlocked = false;
+            playbackListener.unblock();
         }
     }
 
+    /*
+    * Responds to a SELECT event.
+    * When a change occur, the manager prepares by loading more.
+    * If the current item has not been fully loaded,
+    * */
+    private void onSelect() {
+        if (isCurrentIndexLoaded()) {
+            sync();
+        } else if (!isBlocked) {
+            playbackListener.block();
+        }
+
+        load();
+    }
+
     private void sync() {
         final Consumer<StreamInfo> onSuccess = new Consumer<StreamInfo>() {
             @Override
@@ -121,6 +254,47 @@ class MediaSourceManager {
         }
     }
 
+    private void init() {
+        final PlayQueueItem init = playQueue.getCurrent();
+
+        init.getStream().subscribe(new MaybeObserver<StreamInfo>() {
+            @Override
+            public void onSubscribe(@NonNull Disposable d) {
+                if (disposables != null) {
+                    disposables.add(d);
+                } else {
+                    d.dispose();
+                }
+            }
+
+            @Override
+            public void onSuccess(@NonNull StreamInfo streamInfo) {
+                final MediaSource source = playbackListener.sourceOf(streamInfo);
+                insert(playQueue.indexOf(init), source);
+
+                if (getCurrentSourceIndex() != -1) {
+                    playbackListener.init();
+                    sync();
+                    load();
+                } else {
+                    init();
+                }
+            }
+
+            @Override
+            public void onError(@NonNull Throwable e) {
+                playQueue.remove(playQueue.indexOf(init));
+                init();
+            }
+
+            @Override
+            public void onComplete() {
+                playQueue.remove(playQueue.indexOf(init));
+                init();
+            }
+        });
+    }
+
     private void load(final PlayQueueItem item) {
         item.getStream().subscribe(new MaybeObserver<StreamInfo>() {
             @Override
@@ -136,21 +310,38 @@ class MediaSourceManager {
             public void onSuccess(@NonNull StreamInfo streamInfo) {
                 final MediaSource source = playbackListener.sourceOf(streamInfo);
                 insert(playQueue.indexOf(item), source);
-                if (getCurrentSourceIndex() != -1) playbackListener.unblock();
+                tryUnblock();
             }
 
             @Override
             public void onError(@NonNull Throwable e) {
                 playQueue.remove(playQueue.indexOf(item));
+                load();
             }
 
             @Override
             public void onComplete() {
                 playQueue.remove(playQueue.indexOf(item));
+                load();
             }
         });
     }
 
+    /*//////////////////////////////////////////////////////////////////////////
+    // Media Source List Manipulation
+    //////////////////////////////////////////////////////////////////////////*/
+
+    public void replace(final int queueIndex, final MediaSource source) {
+        if (queueIndex < 0) return;
+
+        final int sourceIndex = sourceToQueueIndex.indexOf(queueIndex);
+        if (sourceIndex != -1) {
+            // Add the source after the one to remove, so the window will remain the same in the player
+            sources.addMediaSource(sourceIndex + 1, source);
+            sources.removeMediaSource(sourceIndex);
+        }
+    }
+
     // Insert source into playlist with position in respect to the play queue
     // If the play queue index already exists, then the insert is ignored
     private void insert(final int queueIndex, final MediaSource source) {
@@ -178,17 +369,6 @@ class MediaSourceManager {
         }
     }
 
-    public void replace(final int queueIndex, final MediaSource source) {
-        if (queueIndex < 0) return;
-
-        final int sourceIndex = sourceToQueueIndex.indexOf(queueIndex);
-        if (sourceIndex != -1) {
-            // Add the source after the one to remove, so the window will remain the same in the player
-            sources.addMediaSource(sourceIndex + 1, source);
-            sources.removeMediaSource(sourceIndex);
-        }
-    }
-
     private void swap(final int source, final int target) {
         final int sourceIndex = sourceToQueueIndex.indexOf(source);
         final int targetIndex = sourceToQueueIndex.indexOf(target);
@@ -201,69 +381,4 @@ class MediaSourceManager {
             remove(targetIndex);
         }
     }
-
-    private Subscriber<PlayQueueMessage> getReactor() {
-        return new Subscriber<PlayQueueMessage>() {
-            @Override
-            public void onSubscribe(@NonNull Subscription d) {
-                if (playQueueReactor != null) playQueueReactor.cancel();
-                playQueueReactor = d;
-                playQueueReactor.request(1);
-            }
-
-            @Override
-            public void onNext(@NonNull PlayQueueMessage event) {
-                if (playQueue.size() - playQueue.getIndex() < WINDOW_SIZE && !playQueue.isComplete()) {
-                    playbackListener.block();
-                    playQueue.fetch();
-                }
-
-                // why no pattern matching in Java =(
-                switch (event.type()) {
-                    case INIT:
-                        playbackListener.init();
-                    case APPEND:
-                        load();
-                        break;
-                    case SELECT:
-                        select();
-                        break;
-
-                    case REMOVE:
-                        final RemoveEvent removeEvent = (RemoveEvent) event;
-                        remove(removeEvent.index());
-                        break;
-
-                    case SWAP:
-                        final SwapEvent swapEvent = (SwapEvent) event;
-                        swap(swapEvent.getFrom(), swapEvent.getTo());
-                        break;
-                    case NEXT:
-                        break;
-                    default:
-                        break;
-                }
-
-                if (playQueueReactor != null) playQueueReactor.request(1);
-            }
-
-            @Override
-            public void onError(@NonNull Throwable e) {}
-
-            @Override
-            public void onComplete() {
-                dispose();
-            }
-        };
-    }
-
-    void dispose() {
-        if (loadingReactor != null) loadingReactor.cancel();
-        if (playQueueReactor != null) playQueueReactor.cancel();
-        if (disposables != null) disposables.dispose();
-
-        loadingReactor = null;
-        playQueueReactor = null;
-        disposables = null;
-    }
 }

app/src/main/java/org/schabi/newpipe/player/VideoPlayer.java

@@ -201,26 +201,27 @@ public abstract class VideoPlayer extends BasePlayer implements SimpleExoPlayer.
         simpleExoPlayer.setVideoListener(this);
     }
 
+//    @SuppressWarnings("unchecked")
+//    public void handleIntent2(Intent intent) {
+//        super.handleIntent(intent);
+//        if (DEBUG) Log.d(TAG, "handleIntent() called with: intent = [" + intent + "]");
+//        if (intent == null) return;
+//
+//        selectedIndexStream = intent.getIntExtra(INDEX_SEL_VIDEO_STREAM, -1);
+//
+//        Serializable serializable = intent.getSerializableExtra(VIDEO_STREAMS_LIST);
+//
+//        if (serializable instanceof ArrayList) videoStreamsList = (ArrayList<VideoStream>) serializable;
+//        if (serializable instanceof Vector) videoStreamsList = new ArrayList<>((List<VideoStream>) serializable);
+//
+//        Serializable audioStream = intent.getSerializableExtra(VIDEO_ONLY_AUDIO_STREAM);
+//        if (audioStream != null) videoOnlyAudioStream = (AudioStream) audioStream;
+//
+//        startedFromNewPipe = intent.getBooleanExtra(STARTED_FROM_NEWPIPE, true);
+//        play(true);
+//    }
+
     @SuppressWarnings("unchecked")
-    public void handleIntent2(Intent intent) {
-        super.handleIntent(intent);
-        if (DEBUG) Log.d(TAG, "handleIntent() called with: intent = [" + intent + "]");
-        if (intent == null) return;
-
-        selectedIndexStream = intent.getIntExtra(INDEX_SEL_VIDEO_STREAM, -1);
-
-        Serializable serializable = intent.getSerializableExtra(VIDEO_STREAMS_LIST);
-
-        if (serializable instanceof ArrayList) videoStreamsList = (ArrayList<VideoStream>) serializable;
-        if (serializable instanceof Vector) videoStreamsList = new ArrayList<>((List<VideoStream>) serializable);
-
-        Serializable audioStream = intent.getSerializableExtra(VIDEO_ONLY_AUDIO_STREAM);
-        if (audioStream != null) videoOnlyAudioStream = (AudioStream) audioStream;
-
-        startedFromNewPipe = intent.getBooleanExtra(STARTED_FROM_NEWPIPE, true);
-        play(true);
-    }
-
     public void handleIntent(Intent intent) {
         if (intent == null) return;
 
@@ -454,6 +455,9 @@ public abstract class VideoPlayer extends BasePlayer implements SimpleExoPlayer.
     public void onUpdateProgress(int currentProgress, int duration, int bufferPercent) {
         if (!isPrepared) return;
 
+        if (duration != playbackSeekBar.getMax()) {
+            playbackEndTime.setText(getTimeString(duration));
+        }
         if (currentState != STATE_PAUSED) {
             if (currentState != STATE_PAUSED_SEEK) playbackSeekBar.setProgress(currentProgress);
             playbackCurrentTime.setText(getTimeString(currentProgress));

app/src/main/java/org/schabi/newpipe/playlist/ExternalPlayQueue.java

@@ -32,12 +32,15 @@ public class ExternalPlayQueue extends PlayQueue {
 
     public ExternalPlayQueue(final String playlistUrl,
                              final PlayListInfo info,
-                             final int nextPage,
+                             final int currentPage,
                              final int index) {
         super(index, extractPlaylistItems(info));
 
         this.service = getService(info.service_id);
-        this.pageNumber = new AtomicInteger(nextPage);
+
+        this.isComplete = !info.hasNextPage;
+        this.pageNumber = new AtomicInteger(currentPage + 1);
+
         this.playlistUrl = playlistUrl;
     }
 
@@ -54,6 +57,7 @@ public class ExternalPlayQueue extends PlayQueue {
 
     @Override
     public void fetch() {
+        if (isComplete) return;
         if (fetchReactor != null && !fetchReactor.isDisposed()) return;
 
         final Callable<PlayListInfo> task = new Callable<PlayListInfo>() {
@@ -77,7 +81,6 @@ public class ExternalPlayQueue extends PlayQueue {
         fetchReactor = Maybe.fromCallable(task)
                 .subscribeOn(Schedulers.io())
                 .observeOn(AndroidSchedulers.mainThread())
-                .onErrorComplete()
                 .subscribe(onSuccess);
     }
 

app/src/main/java/org/schabi/newpipe/playlist/PlayQueue.java

@@ -45,7 +45,7 @@ public abstract class PlayQueue {
         streams = Collections.synchronizedList(new ArrayList<PlayQueueItem>());
         streams.addAll(startWith);
 
-        queueIndex = new AtomicInteger(index);
+        queueIndex = new AtomicInteger(97);
 
         eventBroadcast = BehaviorSubject.create();
         broadcastReceiver = eventBroadcast
@@ -62,8 +62,6 @@ public abstract class PlayQueue {
     // load partial queue in the background, does nothing if the queue is complete
     public abstract void fetch();
 
-    // returns a Rx Future to the stream info of the play queue item at index
-    // may return an empty of the queue is incomplete
     public abstract PlayQueueItem get(int index);
 
     public void dispose() {

app/src/main/java/org/schabi/newpipe/playlist/PlayQueueItem.java

@@ -103,6 +103,7 @@ public class PlayQueueItem {
                 .observeOn(AndroidSchedulers.mainThread())
                 .doOnError(onError)
                 .doOnComplete(onComplete)
+                .retry(3)
                 .cache();
     }