-Added full play queue buffering playback manager.

app/src/main/java/org/schabi/newpipe/fragments/playlist/PlaylistFragment.java

@@ -463,7 +463,7 @@ public class PlaylistFragment extends BaseFragment {
     private void handlePlayListInfo(PlayListInfo info, boolean onlyVideos, boolean addVideos) {
         if (currentPlaylistInfo == null) {
             currentPlaylistInfo = info;
-        } else {
+        } else if (currentPlaylistInfo != info) {
             currentPlaylistInfo.related_streams.addAll(info.related_streams);
         }
 

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

@@ -72,7 +72,6 @@ import org.schabi.newpipe.Downloader;
 import org.schabi.newpipe.R;
 import org.schabi.newpipe.extractor.stream_info.StreamInfo;
 import org.schabi.newpipe.playlist.PlayQueue;
-import org.schabi.newpipe.util.Utils;
 
 import java.io.File;
 import java.text.DecimalFormat;
@@ -88,8 +87,9 @@ import java.util.concurrent.atomic.AtomicBoolean;
  */
 @SuppressWarnings({"WeakerAccess", "unused"})
 public abstract class BasePlayer implements Player.EventListener,
-        AudioManager.OnAudioFocusChangeListener, PlaybackManager.PlaybackListener {
+        AudioManager.OnAudioFocusChangeListener, MediaSourceManager.PlaybackListener {
     // TODO: Check api version for deprecated audio manager methods
+
     public static final boolean DEBUG = false;
     public static final String TAG = "BasePlayer";
 
@@ -122,7 +122,7 @@ public abstract class BasePlayer implements Player.EventListener,
     // Playlist
     //////////////////////////////////////////////////////////////////////////*/
 
-    protected PlaybackManager playbackManager;
+    protected MediaSourceManager playbackManager;
     protected PlayQueue playQueue;
 
     /*//////////////////////////////////////////////////////////////////////////
@@ -259,10 +259,9 @@ public abstract class BasePlayer implements Player.EventListener,
 
         isPrepared = false;
 
-        if (simpleExoPlayer.getPlaybackState() != Player.STATE_IDLE) simpleExoPlayer.setPlayWhenReady(false);//simpleExoPlayer.stop();
+        if (simpleExoPlayer.getPlaybackState() != Player.STATE_IDLE) simpleExoPlayer.stop();
         if (videoStartPos > 0) simpleExoPlayer.seekTo(videoStartPos);
-        if (!playbackManager.prepared) simpleExoPlayer.prepare(mediaSource);
+        simpleExoPlayer.prepare(mediaSource);
-        playbackManager.prepared = true;
         simpleExoPlayer.setPlayWhenReady(autoPlay);
     }
 
@@ -549,33 +548,58 @@ public abstract class BasePlayer implements Player.EventListener,
     @Override
     public void onPositionDiscontinuity() {
         int newIndex = simpleExoPlayer.getCurrentWindowIndex();
-        playbackManager.refreshMedia(newIndex);
+        playbackManager.refresh(newIndex);
     }
 
     /*//////////////////////////////////////////////////////////////////////////
     // Playback Listener
     //////////////////////////////////////////////////////////////////////////*/
 
+    private int windowIndex;
+    private long windowPos;
+
     @Override
     public void block() {
-        if (currentState != STATE_BUFFERING) changeState(STATE_BUFFERING);
+        if (currentState != STATE_LOADING) return;
-        simpleExoPlayer.stop();
+
+        changeState(STATE_LOADING);
+        simpleExoPlayer.setPlayWhenReady(false);
+        windowIndex = simpleExoPlayer.getCurrentWindowIndex();
+        windowPos = Math.max(0, simpleExoPlayer.getContentPosition());
     }
 
     @Override
     public void unblock() {
-        if (currentState != STATE_PLAYING) changeState(STATE_PLAYING);
+        if (currentState == STATE_PLAYING) return;
+
+        if (playbackManager.getMediaSource().getSize() > 0) {
+            simpleExoPlayer.seekToDefaultPosition();
+            //simpleExoPlayer.seekTo(windowIndex, windowPos);
+            simpleExoPlayer.setPlayWhenReady(true);
+            changeState(STATE_PLAYING);
+        }
     }
 
     @Override
-    public void resync() {
+    public void sync(final int windowIndex, final long windowPos, final StreamInfo info) {
-        simpleExoPlayer.seekTo(0, 0L);
+        videoUrl = info.webpage_url;
-    }
+        videoThumbnailUrl = info.thumbnail_url;
-
-    @Override
-    public void sync(final StreamInfo info) {
         videoTitle = info.title;
         channelName = info.uploader;
+
+        if (simpleExoPlayer.getCurrentWindowIndex() != windowIndex) {
+            simpleExoPlayer.seekTo(windowIndex, windowPos);
+        } else {
+            simpleExoPlayer.seekTo(windowPos);
+        }
+    }
+
+    @Override
+    public void init() {
+        if (simpleExoPlayer.getPlaybackState() != Player.STATE_IDLE) simpleExoPlayer.stop();
+        simpleExoPlayer.prepare(playbackManager.getMediaSource());
+        simpleExoPlayer.setPlayWhenReady(false);
+        changeState(STATE_BUFFERING);
     }
 
     @Override

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

@@ -229,10 +229,12 @@ public class MainVideoPlayer extends Activity {
         }
 
         @Override
-        public void sync(final StreamInfo info) {
+        public void sync(final int windowIndex, final long windowPos, final StreamInfo info) {
-            super.sync(info);
+            super.sync(windowIndex, windowPos, info);
             titleTextView.setText(getVideoTitle());
             channelTextView.setText(getChannelName());
+
+            playPauseButton.setImageResource(R.drawable.ic_pause_white);
         }
 
         @Override

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

@@ -1,5 +1,7 @@
 package org.schabi.newpipe.player;
 
+import android.util.Log;
+
 import com.google.android.exoplayer2.source.DynamicConcatenatingMediaSource;
 import com.google.android.exoplayer2.source.MediaSource;
 
@@ -7,43 +9,197 @@ import org.reactivestreams.Subscriber;
 import org.reactivestreams.Subscription;
 import org.schabi.newpipe.extractor.stream_info.StreamInfo;
 import org.schabi.newpipe.playlist.PlayQueue;
+import org.schabi.newpipe.playlist.PlayQueueItem;
 import org.schabi.newpipe.playlist.events.PlayQueueMessage;
+import org.schabi.newpipe.playlist.events.RemoveEvent;
+import org.schabi.newpipe.playlist.events.SwapEvent;
 
 import java.util.ArrayList;
 import java.util.Collections;
 import java.util.List;
 
+import io.reactivex.MaybeObserver;
+import io.reactivex.android.schedulers.AndroidSchedulers;
 import io.reactivex.annotations.NonNull;
+import io.reactivex.disposables.CompositeDisposable;
+import io.reactivex.disposables.Disposable;
+import io.reactivex.functions.Consumer;
 
-public class MediaSourceManager {
+class MediaSourceManager {
-    private DynamicConcatenatingMediaSource sources;
+    private final String TAG = "MediaSourceManager@" + Integer.toHexString(hashCode());
-    // indices maps media source index to play queue index
+    private static final int WINDOW_SIZE = 3;
-    // Invariant 1: all indices occur once only in this list
-    private List<Integer> indices;
 
-    private PlaybackListener playbackListener;
+    private final DynamicConcatenatingMediaSource sources;
+    // sourceToQueueIndex maps media source index to play queue index
+    // Invariant 1: this list is sorted in ascending order
+    // Invariant 2: this list contains no duplicates
+    private final List<Integer> sourceToQueueIndex;
+
+    private final PlaybackListener playbackListener;
+    private final PlayQueue playQueue;
 
-    private PlayQueue playQueue;
     private Subscription playQueueReactor;
+    private Subscription loadingReactor;
+    private CompositeDisposable disposables;
 
     interface PlaybackListener {
+        void init();
+
         void block();
         void unblock();
 
-        void resync();
+        void sync(final int windowIndex, final long windowPos, final StreamInfo info);
-        void sync(final StreamInfo info);
         MediaSource sourceOf(final StreamInfo info);
     }
 
-    public MediaSourceManager(@NonNull final MediaSourceManager.PlaybackListener listener,
+    MediaSourceManager(@NonNull final MediaSourceManager.PlaybackListener listener,
-                              @NonNull final PlayQueue playQueue) {
+                       @NonNull final PlayQueue playQueue) {
         this.sources = new DynamicConcatenatingMediaSource();
-        this.indices = Collections.synchronizedList(new ArrayList<Integer>());
+        this.sourceToQueueIndex = Collections.synchronizedList(new ArrayList<Integer>());
 
         this.playbackListener = listener;
         this.playQueue = playQueue;
 
-        playQueue.getEventBroadcast().subscribe(getReactor());
+        disposables = new CompositeDisposable();
+
+        playQueue.getBroadcastReceiver()
+                .observeOn(AndroidSchedulers.mainThread())
+                .subscribe(getReactor());
+    }
+
+    int getCurrentSourceIndex() {
+        return sourceToQueueIndex.indexOf(playQueue.getIndex());
+    }
+
+    @NonNull
+    DynamicConcatenatingMediaSource getMediaSource() {
+        return sources;
+    }
+
+    void refresh(final int newSourceIndex) {
+        if (newSourceIndex == getCurrentSourceIndex()) return;
+
+        if (newSourceIndex == getCurrentSourceIndex() + 1) {
+            playQueue.incrementIndex();
+        } else {
+            //something went wrong
+            Log.e(TAG, "Refresh media failed, reloading.");
+        }
+
+        sync();
+    }
+
+    private void select() {
+        if (getCurrentSourceIndex() != -1) {
+            sync();
+        } else {
+            playbackListener.block();
+            load();
+        }
+    }
+
+    private void sync() {
+        final Consumer<StreamInfo> onSuccess = new Consumer<StreamInfo>() {
+            @Override
+            public void accept(StreamInfo streamInfo) throws Exception {
+                playbackListener.sync(getCurrentSourceIndex(), 0L, streamInfo);
+            }
+        };
+
+        playQueue.getCurrent().getStream().subscribe(onSuccess);
+    }
+
+    private void load() {
+        final int currentIndex = playQueue.getIndex();
+        load(playQueue.get(currentIndex));
+
+        final int leftBound = Math.max(0, currentIndex - WINDOW_SIZE);
+        final int rightBound = Math.min(playQueue.size(), currentIndex + WINDOW_SIZE);
+        final List<PlayQueueItem> items = playQueue.getStreams().subList(leftBound, rightBound);
+        for (final PlayQueueItem item: items) {
+            load(item);
+        }
+    }
+
+    private void load(final PlayQueueItem item) {
+        item.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(item), source);
+                if (getCurrentSourceIndex() != -1) playbackListener.unblock();
+            }
+
+            @Override
+            public void onError(@NonNull Throwable e) {
+                playQueue.remove(playQueue.indexOf(item));
+            }
+
+            @Override
+            public void onComplete() {
+                playQueue.remove(playQueue.indexOf(item));
+            }
+        });
+    }
+
+    // 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) {
+        if (queueIndex < 0) return;
+
+        int pos = Collections.binarySearch(sourceToQueueIndex, queueIndex);
+        if (pos < 0) {
+            final int sourceIndex = -pos-1;
+            sourceToQueueIndex.add(sourceIndex, queueIndex);
+            sources.addMediaSource(sourceIndex, source);
+        }
+    }
+
+    private void remove(final int queueIndex) {
+        if (queueIndex < 0) return;
+
+        final int sourceIndex = sourceToQueueIndex.indexOf(queueIndex);
+        if (sourceIndex != -1) {
+            sourceToQueueIndex.remove(sourceIndex);
+            sources.removeMediaSource(sourceIndex);
+            // Will be slow on really large arrays, fast enough for typical use case
+            for (int i = sourceIndex; i < sourceToQueueIndex.size(); i++) {
+                sourceToQueueIndex.set(i, sourceToQueueIndex.get(i) - 1);
+            }
+        }
+    }
+
+    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);
+
+        if (sourceIndex != -1 && targetIndex != -1) {
+            sources.moveMediaSource(sourceIndex, targetIndex);
+        } else if (sourceIndex != -1) {
+            remove(sourceIndex);
+        } else if (targetIndex != -1) {
+            remove(targetIndex);
+        }
     }
 
     private Subscriber<PlayQueueMessage> getReactor() {
@@ -57,18 +213,33 @@ public class MediaSourceManager {
 
             @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:
-                        break;
+                        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;
                 }
@@ -77,9 +248,7 @@ public class MediaSourceManager {
             }
 
             @Override
-            public void onError(@NonNull Throwable e) {
+            public void onError(@NonNull Throwable e) {}
-
-            }
 
             @Override
             public void onComplete() {
@@ -88,8 +257,13 @@ public class MediaSourceManager {
         };
     }
 
-    public void 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/PlaybackManager.java

@@ -9,7 +9,6 @@ import org.reactivestreams.Subscriber;
 import org.reactivestreams.Subscription;
 import org.schabi.newpipe.extractor.stream_info.StreamInfo;
 import org.schabi.newpipe.playlist.PlayQueue;
-import org.schabi.newpipe.playlist.events.PlayQueueEvent;
 import org.schabi.newpipe.playlist.PlayQueueItem;
 import org.schabi.newpipe.playlist.events.PlayQueueMessage;
 
@@ -55,7 +54,7 @@ public class PlaybackManager {
         this.listener = listener;
         this.playQueue = playQueue;
 
-        playQueue.getEventBroadcast().subscribe(getReactor());
+        playQueue.getBroadcastReceiver().subscribe(getReactor());
     }
 
     @NonNull
@@ -65,7 +64,9 @@ public class PlaybackManager {
 
     private void reload() {
         listener.block();
-        load(0);
+        mediaSource = new DynamicConcatenatingMediaSource();
+        syncInfos.clear();
+        load();
     }
 
     public void changeSource(final MediaSource newSource) {
@@ -87,11 +88,6 @@ public class PlaybackManager {
         }
     }
 
-    private void removeCurrent() {
-        mediaSource.removeMediaSource(0);
-        syncInfos.remove(0);
-    }
-
     private Subscription loaderReactor;
 
     private void load() {
@@ -152,11 +148,6 @@ public class PlaybackManager {
         if (mediaSource.getSize() > 0) listener.unblock();
     }
 
-    private void init() {
-        listener.block();
-        load();
-    }
-
     private void clear(int from) {
         while (mediaSource.getSize() > from) {
             mediaSource.removeMediaSource(from);
@@ -181,15 +172,13 @@ public class PlaybackManager {
                 }
 
                 switch (event.type()) {
+                    case SELECT:
                     case INIT:
-                        init();
+                        reload();
                         break;
                     case APPEND:
                         load();
                         break;
-                    case SELECT:
-                        reload();
-                        break;
                     case REMOVE:
                     case SWAP:
                         load(1);

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

@@ -221,26 +221,6 @@ public abstract class VideoPlayer extends BasePlayer implements SimpleExoPlayer.
         play(true);
     }
 
-    @Override
-    public MediaSource sourceOf(final StreamInfo info) {
-        videoStreamsList = Utils.getSortedStreamVideosList(context, info.video_streams, info.video_only_streams, false);
-        videoOnlyAudioStream = Utils.getHighestQualityAudio(info.audio_streams);
-
-        return buildMediaSource(getSelectedVideoStream().url, MediaFormat.getSuffixById(getSelectedVideoStream().format));
-    }
-
-    @Override
-    public void block() {
-        if (currentState != STATE_BUFFERING) changeState(STATE_BUFFERING);
-        simpleExoPlayer.stop();
-    }
-
-    @Override
-    public void unblock() {
-        if (currentState != STATE_PLAYING) changeState(STATE_PLAYING);
-        if (!isPlaying()) play(true);
-    }
-
     public void handleIntent(Intent intent) {
         if (intent == null) return;
 
@@ -256,14 +236,43 @@ public abstract class VideoPlayer extends BasePlayer implements SimpleExoPlayer.
         else return;
 
         playQueue = new ExternalPlayQueue(url, info, nextPage, index);
-        playbackManager = new PlaybackManager(this, playQueue);
+        playbackManager = new MediaSourceManager(this, playQueue);
-        mediaSource = playbackManager.getMediaSource();
     }
 
     public void play(boolean autoPlay) {
         playUrl(getSelectedVideoStream().url, MediaFormat.getSuffixById(getSelectedVideoStream().format), autoPlay);
     }
 
+    @Override
+    public void sync(final int windowIndex, final long windowPos, final StreamInfo info) {
+        super.sync(windowIndex, windowPos, info);
+
+        qualityPopupMenu.getMenu().removeGroup(qualityPopupMenuGroupId);
+        for (int i = 0; i < info.video_streams.size(); i++) {
+            VideoStream videoStream = info.video_streams.get(i);
+            qualityPopupMenu.getMenu().add(qualityPopupMenuGroupId, i, Menu.NONE, MediaFormat.getNameById(videoStream.format) + " " + videoStream.resolution);
+        }
+        qualityTextView.setText(info.video_streams.get(selectedIndexStream).resolution);
+        qualityPopupMenu.setOnMenuItemClickListener(this);
+        qualityPopupMenu.setOnDismissListener(this);
+
+        playbackSpeedPopupMenu.getMenu().removeGroup(playbackSpeedPopupMenuGroupId);
+        buildPlaybackSpeedMenu(playbackSpeedPopupMenu);
+    }
+
+    @Override
+    public MediaSource sourceOf(final StreamInfo info) {
+        final List<VideoStream> videos = Utils.getSortedStreamVideosList(context, info.video_streams, info.video_only_streams, false);
+        final VideoStream video = videos.get(Utils.getDefaultResolution(context, videos));
+
+        final MediaSource mediaSource = super.buildMediaSource(video.url, MediaFormat.getSuffixById(video.format));
+        if (!video.isVideoOnly) return mediaSource;
+
+        final AudioStream audio = Utils.getHighestQualityAudio(info.audio_streams);
+        final Uri audioUri = Uri.parse(audio.url);
+        return new MergingMediaSource(mediaSource, new ExtractorMediaSource(audioUri, cacheDataSourceFactory, extractorsFactory, null, null));
+    }
+
     @Override
     public void playUrl(String url, String format, boolean autoPlay) {
         if (DEBUG) Log.d(TAG, "play() called with: url = [" + url + "], autoPlay = [" + autoPlay + "]");

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

@@ -11,7 +11,6 @@ import org.schabi.newpipe.extractor.exceptions.ExtractionException;
 import org.schabi.newpipe.playlist.events.AppendEvent;
 import org.schabi.newpipe.playlist.events.InitEvent;
 import org.schabi.newpipe.playlist.events.NextEvent;
-import org.schabi.newpipe.playlist.events.PlayQueueEvent;
 import org.schabi.newpipe.playlist.events.PlayQueueMessage;
 import org.schabi.newpipe.playlist.events.RemoveEvent;
 import org.schabi.newpipe.playlist.events.SelectEvent;
@@ -31,11 +30,11 @@ public abstract class PlayQueue {
     private final String TAG = "PlayQueue@" + Integer.toHexString(hashCode());
     public static final boolean DEBUG = true;
 
-    private List<PlayQueueItem> streams;
+    private final List<PlayQueueItem> streams;
-    private AtomicInteger queueIndex;
+    private final AtomicInteger queueIndex;
 
-    private BehaviorSubject<PlayQueueMessage> eventBus;
+    private final BehaviorSubject<PlayQueueMessage> eventBroadcast;
-    private Flowable<PlayQueueMessage> eventBroadcast;
+    private final Flowable<PlayQueueMessage> broadcastReceiver;
     private Subscription reportingReactor;
 
     PlayQueue() {
@@ -48,13 +47,12 @@ public abstract class PlayQueue {
 
         queueIndex = new AtomicInteger(index);
 
-        eventBus = BehaviorSubject.create();
+        eventBroadcast = BehaviorSubject.create();
-        eventBroadcast = eventBus
+        broadcastReceiver = eventBroadcast
                 .startWith(new InitEvent())
-                .replay(20)
                 .toFlowable(BackpressureStrategy.BUFFER);
 
-        if (DEBUG) eventBroadcast.subscribe(getSelfReporter());
+        if (DEBUG) broadcastReceiver.subscribe(getSelfReporter());
     }
 
     // a queue is complete if it has loaded all items in an external playlist
@@ -69,10 +67,16 @@ public abstract class PlayQueue {
     public abstract PlayQueueItem get(int index);
 
     public void dispose() {
+        eventBroadcast.onComplete();
+
         if (reportingReactor != null) reportingReactor.cancel();
         reportingReactor = null;
     }
 
+    public PlayQueueItem getCurrent() {
+        return streams.get(getIndex());
+    }
+
     public int size() {
         return streams.size();
     }
@@ -83,12 +87,18 @@ public abstract class PlayQueue {
     }
 
     @NonNull
-    public Flowable<PlayQueueMessage> getEventBroadcast() {
+    public Flowable<PlayQueueMessage> getBroadcastReceiver() {
-        return eventBroadcast;
+        return broadcastReceiver;
     }
 
     private void broadcast(final PlayQueueMessage event) {
-        eventBus.onNext(event);
+        eventBroadcast.onNext(event);
+    }
+
+    public int indexOf(final PlayQueueItem item) {
+        // reference equality, can't think of a better way to do this
+        // todo: better than this
+        return streams.indexOf(item);
     }
 
     public int getIndex() {
@@ -96,7 +106,7 @@ public abstract class PlayQueue {
     }
 
     public void setIndex(final int index) {
-        queueIndex.set(Math.max(0, index));
+        queueIndex.set(Math.min(Math.max(0, index), streams.size() - 1));
         broadcast(new SelectEvent(index));
     }
 

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

@@ -8,12 +8,13 @@ import android.view.ViewGroup;
 
 import org.schabi.newpipe.R;
 import org.schabi.newpipe.info_list.StreamInfoItemHolder;
-import org.schabi.newpipe.playlist.events.PlayQueueEvent;
+import org.schabi.newpipe.playlist.events.PlayQueueMessage;
 
 import java.util.List;
 
+import io.reactivex.Observer;
+import io.reactivex.annotations.NonNull;
 import io.reactivex.disposables.Disposable;
-import io.reactivex.functions.Consumer;
 
 /**
  * Created by Christian Schabesberger on 01.08.16.
@@ -63,7 +64,7 @@ public class PlayQueueAdapter extends RecyclerView.Adapter<RecyclerView.ViewHold
         this.playQueueItemBuilder = new PlayQueueItemBuilder();
         this.playQueue = playQueue;
 
-        playQueueReactor = getReactor();
+        startReactor();
     }
 
     public void setSelectedListener(final PlayQueueItemBuilder.OnSelectedListener listener) {
@@ -86,21 +87,36 @@ public class PlayQueueAdapter extends RecyclerView.Adapter<RecyclerView.ViewHold
         playQueue.swap(source, target);
     }
 
-    private Disposable getReactor() {
+    private void startReactor() {
-        final Consumer<PlayQueueEvent> onNext = new Consumer<PlayQueueEvent>() {
+        final Observer<PlayQueueMessage> observer = new Observer<PlayQueueMessage>() {
             @Override
-            public void accept(PlayQueueEvent playQueueEvent) throws Exception {
+            public void onSubscribe(@NonNull Disposable d) {
+                if (playQueueReactor != null) playQueueReactor.dispose();
+                playQueueReactor = d;
+            }
+
+            @Override
+            public void onNext(@NonNull PlayQueueMessage playQueueMessage) {
                 notifyDataSetChanged();
             }
+
+            @Override
+            public void onError(@NonNull Throwable e) {}
+
+            @Override
+            public void onComplete() {
+                dispose();
+            }
         };
 
-        return playQueue.getEventBroadcast()
+        playQueue.getBroadcastReceiver()
                 .toObservable()
-                .subscribe(onNext);
+                .subscribe(observer);
     }
 
     public void dispose() {
         if (playQueueReactor != null) playQueueReactor.dispose();
+        playQueueReactor = null;
     }
 
     public void setHeader(View header) {

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

@@ -4,6 +4,8 @@ import android.support.annotation.NonNull;
 import android.support.annotation.Nullable;
 
 import org.schabi.newpipe.extractor.NewPipe;
+import org.schabi.newpipe.extractor.StreamingService;
+import org.schabi.newpipe.extractor.exceptions.ExtractionException;
 import org.schabi.newpipe.extractor.stream_info.StreamExtractor;
 import org.schabi.newpipe.extractor.stream_info.StreamInfo;
 import org.schabi.newpipe.extractor.stream_info.StreamInfoItem;
@@ -18,16 +20,16 @@ import io.reactivex.schedulers.Schedulers;
 
 public class PlayQueueItem {
 
-    private String title;
+    final private String title;
-    private String url;
+    final private String url;
-    private int serviceId;
+    final private int serviceId;
-    private int duration;
+    final private int duration;
 
     private boolean isDone;
     private Throwable error;
     private Maybe<StreamInfo> stream;
 
-    public PlayQueueItem(final StreamInfoItem streamInfoItem) {
+    PlayQueueItem(final StreamInfoItem streamInfoItem) {
         this.title = streamInfoItem.getTitle();
         this.url = streamInfoItem.getLink();
         this.serviceId = streamInfoItem.service_id;
@@ -71,10 +73,13 @@ public class PlayQueueItem {
 
     @NonNull
     private Maybe<StreamInfo> getInfo() {
+        final StreamingService service = getService(serviceId);
+        if (service == null) return Maybe.empty();
+
         final Callable<StreamInfo> task = new Callable<StreamInfo>() {
             @Override
             public StreamInfo call() throws Exception {
-                final StreamExtractor extractor = NewPipe.getService(serviceId).getExtractorInstance(url);
+                final StreamExtractor extractor = service.getExtractorInstance(url);
                 return StreamInfo.getVideoInfo(extractor);
             }
         };
@@ -100,4 +105,12 @@ public class PlayQueueItem {
                 .doOnComplete(onComplete)
                 .cache();
     }
+
+    private StreamingService getService(final int serviceId) {
+        try {
+            return NewPipe.getService(serviceId);
+        } catch (ExtractionException e) {
+            return null;
+        }
+    }
 }

app/src/main/java/org/schabi/newpipe/playlist/events/AppendEvent.java

@@ -2,7 +2,7 @@ package org.schabi.newpipe.playlist.events;
 
 
 public class AppendEvent implements PlayQueueMessage {
-    private int amount;
+    final private int amount;
 
     @Override
     public PlayQueueEvent type() {

app/src/main/java/org/schabi/newpipe/playlist/events/NextEvent.java

@@ -2,7 +2,7 @@ package org.schabi.newpipe.playlist.events;
 
 
 public class NextEvent implements PlayQueueMessage {
-    private int newIndex;
+    final private int newIndex;
 
     @Override
     public PlayQueueEvent type() {

app/src/main/java/org/schabi/newpipe/playlist/events/RemoveEvent.java

@@ -1,8 +1,8 @@
 package org.schabi.newpipe.playlist.events;
 
 
-public class RemoveEvent extends PlayQueueMessage {
+public class RemoveEvent implements PlayQueueMessage {
-    private int index;
+    final private int index;
 
     @Override
     public PlayQueueEvent type() {

app/src/main/java/org/schabi/newpipe/playlist/events/SelectEvent.java

@@ -2,7 +2,7 @@ package org.schabi.newpipe.playlist.events;
 
 
 public class SelectEvent implements PlayQueueMessage {
-    private int newIndex;
+    final private int newIndex;
 
     @Override
     public PlayQueueEvent type() {

app/src/main/java/org/schabi/newpipe/playlist/events/SwapEvent.java

@@ -2,8 +2,8 @@ package org.schabi.newpipe.playlist.events;
 
 
 public class SwapEvent implements PlayQueueMessage {
-    private int from;
+    final private int from;
-    private int to;
+    final private int to;
 
     @Override
     public PlayQueueEvent type() {

app/src/main/res/layout/activity_main_player.xml

@@ -11,6 +11,7 @@
         android:id="@+id/aspectRatioLayout"
         android:layout_width="match_parent"
         android:layout_height="match_parent"
+        android:layout_centerInParent="true"
         android:layout_gravity="center">
 
         <SurfaceView