-Improved play queue message bus

-Hooking play queue engines with video players (to be removed)
-Proof of concept for previous and next controls
This commit is contained in:
John Zhen M 2017-09-02 11:06:36 -07:00 committed by John Zhen Mo
parent b859823011
commit 74b58cae59
17 changed files with 345 additions and 95 deletions

View File

@ -461,7 +461,11 @@ public class PlaylistFragment extends BaseFragment {
}
private void handlePlayListInfo(PlayListInfo info, boolean onlyVideos, boolean addVideos) {
if (currentPlaylistInfo == null) currentPlaylistInfo = info;
if (currentPlaylistInfo == null) {
currentPlaylistInfo = info;
} else {
currentPlaylistInfo.related_streams.addAll(info.related_streams);
}
animateView(errorPanel, false, 300);
animateView(playlistStreams, true, 200);
@ -494,12 +498,9 @@ public class PlaylistFragment extends BaseFragment {
hasNextPage = info.hasNextPage;
if (!hasNextPage) infoListAdapter.showFooter(false);
//if (!listRestored) {
if (addVideos) {
infoListAdapter.addInfoItemList(info.related_streams);
currentPlaylistInfo.related_streams.addAll(info.related_streams);
}
//}
}
@Override

View File

@ -259,9 +259,10 @@ public abstract class BasePlayer implements Player.EventListener,
isPrepared = false;
if (simpleExoPlayer.getPlaybackState() != Player.STATE_IDLE) simpleExoPlayer.stop();
if (simpleExoPlayer.getPlaybackState() != Player.STATE_IDLE) simpleExoPlayer.setPlayWhenReady(false);//simpleExoPlayer.stop();
if (videoStartPos > 0) simpleExoPlayer.seekTo(videoStartPos);
simpleExoPlayer.prepare(mediaSource);
if (!playbackManager.prepared) simpleExoPlayer.prepare(mediaSource);
playbackManager.prepared = true;
simpleExoPlayer.setPlayWhenReady(autoPlay);
}
@ -557,7 +558,8 @@ public abstract class BasePlayer implements Player.EventListener,
@Override
public void block() {
if (currentState != STATE_LOADING) changeState(STATE_LOADING);
if (currentState != STATE_BUFFERING) changeState(STATE_BUFFERING);
simpleExoPlayer.stop();
}
@Override
@ -565,6 +567,11 @@ public abstract class BasePlayer implements Player.EventListener,
if (currentState != STATE_PLAYING) changeState(STATE_PLAYING);
}
@Override
public void resync() {
simpleExoPlayer.seekTo(0, 0L);
}
@Override
public void sync(final StreamInfo info) {
videoTitle = info.title;

View File

@ -40,6 +40,7 @@ import android.widget.TextView;
import android.widget.Toast;
import org.schabi.newpipe.R;
import org.schabi.newpipe.extractor.stream_info.StreamInfo;
import org.schabi.newpipe.util.AnimationUtils;
import org.schabi.newpipe.util.NavigationHelper;
import org.schabi.newpipe.util.PermissionHelper;
@ -227,6 +228,13 @@ public class MainVideoPlayer extends Activity {
channelTextView.setText(getUploaderName());
}
@Override
public void sync(final StreamInfo info) {
super.sync(info);
titleTextView.setText(getVideoTitle());
channelTextView.setText(getChannelName());
}
@Override
public void playUrl(String url, String format, boolean autoPlay) {
super.playUrl(url, format, autoPlay);

View File

@ -0,0 +1,95 @@
package org.schabi.newpipe.player;
import com.google.android.exoplayer2.source.DynamicConcatenatingMediaSource;
import com.google.android.exoplayer2.source.MediaSource;
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.PlayQueueMessage;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import io.reactivex.annotations.NonNull;
public class MediaSourceManager {
private DynamicConcatenatingMediaSource sources;
// indices maps media source index to play queue index
// Invariant 1: all indices occur once only in this list
private List<Integer> indices;
private PlaybackListener playbackListener;
private PlayQueue playQueue;
private Subscription playQueueReactor;
interface PlaybackListener {
void block();
void unblock();
void resync();
void sync(final StreamInfo info);
MediaSource sourceOf(final StreamInfo info);
}
public MediaSourceManager(@NonNull final MediaSourceManager.PlaybackListener listener,
@NonNull final PlayQueue playQueue) {
this.sources = new DynamicConcatenatingMediaSource();
this.indices = Collections.synchronizedList(new ArrayList<Integer>());
this.playbackListener = listener;
this.playQueue = playQueue;
playQueue.getEventBroadcast().subscribe(getReactor());
}
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) {
switch (event.type()) {
case INIT:
break;
case APPEND:
break;
case SELECT:
break;
case REMOVE:
case SWAP:
break;
case NEXT:
default:
break;
}
if (playQueueReactor != null) playQueueReactor.request(1);
}
@Override
public void onError(@NonNull Throwable e) {
}
@Override
public void onComplete() {
dispose();
}
};
}
public void dispose() {
if (playQueueReactor != null) playQueueReactor.cancel();
playQueueReactor = null;
}
}

View File

@ -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,24 +9,25 @@ 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.PlayQueueEvent;
import org.schabi.newpipe.playlist.events.PlayQueueEvent;
import org.schabi.newpipe.playlist.PlayQueueItem;
import org.schabi.newpipe.playlist.events.PlayQueueMessage;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import io.reactivex.Maybe;
import io.reactivex.android.schedulers.AndroidSchedulers;
import io.reactivex.annotations.NonNull;
import io.reactivex.schedulers.Schedulers;
public class PlaybackManager {
private final String TAG = "PlaybackManager@" + Integer.toHexString(hashCode());
private static final int WINDOW_SIZE = 5;
private static final int WINDOW_SIZE = 3;
private DynamicConcatenatingMediaSource mediaSource;
private List<PlayQueueItem> queueSource;
private List<StreamInfo> syncInfos;
private int sourceIndex;
private PlaybackListener listener;
@ -32,10 +35,13 @@ public class PlaybackManager {
private Subscription playQueueReactor;
public boolean prepared = false;
interface PlaybackListener {
void block();
void unblock();
void resync();
void sync(final StreamInfo info);
MediaSource sourceOf(final StreamInfo info);
}
@ -43,13 +49,13 @@ public class PlaybackManager {
public PlaybackManager(@NonNull final PlaybackListener listener,
@NonNull final PlayQueue playQueue) {
this.mediaSource = new DynamicConcatenatingMediaSource();
this.queueSource = Collections.synchronizedList(new ArrayList<PlayQueueItem>(10));
this.syncInfos = Collections.synchronizedList(new ArrayList<StreamInfo>());
this.sourceIndex = 0;
this.listener = listener;
this.playQueue = playQueue;
playQueue.getPlayQueueFlowable().subscribe(getReactor());
playQueue.getEventBroadcast().subscribe(getReactor());
}
@NonNull
@ -63,10 +69,8 @@ public class PlaybackManager {
}
public void changeSource(final MediaSource newSource) {
listener.block();
this.mediaSource.removeMediaSource(0);
this.mediaSource.addMediaSource(0, newSource);
listener.unblock();
}
public void refreshMedia(final int newMediaIndex) {
@ -75,43 +79,42 @@ public class PlaybackManager {
if (newMediaIndex == sourceIndex + 1) {
playQueue.incrementIndex();
mediaSource.removeMediaSource(0);
queueSource.remove(0);
syncInfos.remove(0);
} else {
//something went wrong
Log.e(TAG, "Refresh media failed, reloading.");
reload();
}
}
private void removeCurrent() {
listener.block();
mediaSource.removeMediaSource(0);
queueSource.remove(0);
listener.unblock();
syncInfos.remove(0);
}
private Subscription loaderReactor;
private void load() {
if (mediaSource.getSize() < WINDOW_SIZE && queueSource.size() < WINDOW_SIZE)
load(mediaSource.getSize());
if (mediaSource.getSize() < WINDOW_SIZE) load(mediaSource.getSize());
}
private void load(final int from) {
clear(from);
if (loaderReactor != null) loaderReactor.cancel();
// Fetch queue items
//todo fix out of bound
final int index = playQueue.getIndex();
List<Maybe<StreamInfo>> maybes = new ArrayList<>();
for (int i = from; i < WINDOW_SIZE; i++) {
final int index = playQueue.getIndex() + i;
final PlayQueueItem item = playQueue.get(index);
if (queueSource.size() > i) queueSource.set(i, item);
else queueSource.add(item);
final PlayQueueItem item = playQueue.get(index + i);
maybes.add(item.getStream());
}
// Stop loading and clear pending media sources
if (loaderReactor != null) loaderReactor.cancel();
clear(from);
// Start sequential loading of media sources
Maybe.concat(maybes).subscribe(getSubscriber());
}
@ -127,13 +130,14 @@ public class PlaybackManager {
@Override
public void onNext(StreamInfo streamInfo) {
mediaSource.addMediaSource(listener.sourceOf(streamInfo));
syncInfos.add(streamInfo);
tryUnblock();
loaderReactor.request(1);
}
@Override
public void onError(Throwable t) {
playQueue.remove(queueSource.size());
playQueue.remove(playQueue.getIndex());
}
@Override
@ -145,7 +149,7 @@ public class PlaybackManager {
}
private void tryUnblock() {
if (mediaSource.getSize() > 0 && queueSource.size() > 0) listener.unblock();
if (mediaSource.getSize() > 0) listener.unblock();
}
private void init() {
@ -155,19 +159,13 @@ public class PlaybackManager {
private void clear(int from) {
while (mediaSource.getSize() > from) {
queueSource.remove(from);
mediaSource.removeMediaSource(from);
syncInfos.remove(from);
}
}
private void clear() {
listener.block();
clear(0);
listener.unblock();
}
private Subscriber<PlayQueueEvent> getReactor() {
return new Subscriber<PlayQueueEvent>() {
private Subscriber<PlayQueueMessage> getReactor() {
return new Subscriber<PlayQueueMessage>() {
@Override
public void onSubscribe(@NonNull Subscription d) {
if (playQueueReactor != null) playQueueReactor.cancel();
@ -176,23 +174,19 @@ public class PlaybackManager {
}
@Override
public void onNext(@NonNull PlayQueueEvent event) {
public void onNext(@NonNull PlayQueueMessage event) {
if (playQueue.getStreams().size() - playQueue.getIndex() < WINDOW_SIZE && !playQueue.isComplete()) {
listener.block();
playQueue.fetch();
}
switch (event) {
switch (event.type()) {
case INIT:
init();
break;
case APPEND:
load();
break;
case REMOVE_CURRENT:
removeCurrent();
load();
break;
case SELECT:
reload();
break;
@ -200,15 +194,13 @@ public class PlaybackManager {
case SWAP:
load(1);
break;
case CLEAR:
clear();
break;
case NEXT:
default:
break;
}
tryUnblock();
if (!syncInfos.isEmpty()) listener.sync(syncInfos.get(0));
if (playQueueReactor != null) playQueueReactor.request(1);
}

View File

@ -229,10 +229,16 @@ public abstract class VideoPlayer extends BasePlayer implements SimpleExoPlayer.
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() {
play(true);
super.unblock();
if (currentState != STATE_PLAYING) changeState(STATE_PLAYING);
if (!isPlaying()) play(true);
}
public void handleIntent(Intent intent) {

View File

@ -83,6 +83,7 @@ public class ExternalPlayQueue extends PlayQueue {
@Override
public void dispose() {
super.dispose();
if (fetchReactor != null) fetchReactor.dispose();
}

View File

@ -1,10 +1,21 @@
package org.schabi.newpipe.playlist;
import android.support.annotation.NonNull;
import android.util.Log;
import org.reactivestreams.Subscriber;
import org.reactivestreams.Subscription;
import org.schabi.newpipe.extractor.NewPipe;
import org.schabi.newpipe.extractor.StreamingService;
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;
import org.schabi.newpipe.playlist.events.SwapEvent;
import java.util.ArrayList;
import java.util.Collection;
@ -18,12 +29,14 @@ import io.reactivex.subjects.BehaviorSubject;
public abstract class PlayQueue {
private final String TAG = "PlayQueue@" + Integer.toHexString(hashCode());
public static final boolean DEBUG = true;
private List<PlayQueueItem> streams;
private AtomicInteger queueIndex;
private BehaviorSubject<PlayQueueEvent> changeBroadcast;
private Flowable<PlayQueueEvent> playQueueFlowable;
private BehaviorSubject<PlayQueueMessage> eventBus;
private Flowable<PlayQueueMessage> eventBroadcast;
private Subscription reportingReactor;
PlayQueue() {
this(0, Collections.<PlayQueueItem>emptyList());
@ -35,8 +48,13 @@ public abstract class PlayQueue {
queueIndex = new AtomicInteger(index);
changeBroadcast = BehaviorSubject.create();
playQueueFlowable = changeBroadcast.startWith(PlayQueueEvent.INIT).toFlowable(BackpressureStrategy.BUFFER);
eventBus = BehaviorSubject.create();
eventBroadcast = eventBus
.startWith(new InitEvent())
.replay(20)
.toFlowable(BackpressureStrategy.BUFFER);
if (DEBUG) eventBroadcast.subscribe(getSelfReporter());
}
// a queue is complete if it has loaded all items in an external playlist
@ -50,7 +68,10 @@ public abstract class PlayQueue {
// may return an empty of the queue is incomplete
public abstract PlayQueueItem get(int index);
public abstract void dispose();
public void dispose() {
if (reportingReactor != null) reportingReactor.cancel();
reportingReactor = null;
}
public int size() {
return streams.size();
@ -62,12 +83,12 @@ public abstract class PlayQueue {
}
@NonNull
public Flowable<PlayQueueEvent> getPlayQueueFlowable() {
return playQueueFlowable;
public Flowable<PlayQueueMessage> getEventBroadcast() {
return eventBroadcast;
}
private void broadcast(final PlayQueueEvent event) {
changeBroadcast.onNext(event);
private void broadcast(final PlayQueueMessage event) {
eventBus.onNext(event);
}
public int getIndex() {
@ -75,43 +96,30 @@ public abstract class PlayQueue {
}
public void setIndex(final int index) {
queueIndex.set(index);
broadcast(PlayQueueEvent.SELECT);
queueIndex.set(Math.max(0, index));
broadcast(new SelectEvent(index));
}
public void incrementIndex() {
queueIndex.incrementAndGet();
broadcast(PlayQueueEvent.NEXT);
final int index = queueIndex.incrementAndGet();
broadcast(new NextEvent(index));
}
protected void append(final PlayQueueItem item) {
streams.add(item);
broadcast(PlayQueueEvent.APPEND);
broadcast(new AppendEvent(1));
}
protected void append(final Collection<PlayQueueItem> items) {
streams.addAll(items);
broadcast(PlayQueueEvent.APPEND);
broadcast(new AppendEvent(items.size()));
}
public void remove(final int index) {
if (index >= streams.size()) return;
final boolean isCurrent = index == queueIndex.get();
streams.remove(index);
if (isCurrent) {
broadcast(PlayQueueEvent.REMOVE_CURRENT);
} else {
broadcast(PlayQueueEvent.REMOVE);
}
}
protected void clear() {
if (!streams.isEmpty()) {
streams.clear();
broadcast(PlayQueueEvent.CLEAR);
}
broadcast(new RemoveEvent(index));
}
protected void swap(final int source, final int target) {
@ -131,7 +139,7 @@ public abstract class PlayQueue {
queueIndex.set(newIndex);
}
broadcast(PlayQueueEvent.SWAP);
broadcast(new SwapEvent(source, target));
}
}
@ -142,5 +150,32 @@ public abstract class PlayQueue {
return null;
}
}
private Subscriber<PlayQueueMessage> getSelfReporter() {
return new Subscriber<PlayQueueMessage>() {
@Override
public void onSubscribe(Subscription s) {
if (reportingReactor != null) reportingReactor.cancel();
reportingReactor = s;
reportingReactor.request(1);
}
@Override
public void onNext(PlayQueueMessage event) {
Log.d(TAG, "Received broadcast: " + event.type().name() + ". Current index: " + getIndex() + ", play queue length: " + size() + ".");
reportingReactor.request(1);
}
@Override
public void onError(Throwable t) {
Log.e(TAG, "Received broadcast error", t);
}
@Override
public void onComplete() {
Log.d(TAG, "Broadcast is shut down.");
}
};
}
}

View File

@ -8,6 +8,7 @@ import android.view.ViewGroup;
import org.schabi.newpipe.R;
import org.schabi.newpipe.info_list.StreamInfoItemHolder;
import org.schabi.newpipe.playlist.events.PlayQueueEvent;
import java.util.List;
@ -85,10 +86,6 @@ public class PlayQueueAdapter extends RecyclerView.Adapter<RecyclerView.ViewHold
playQueue.swap(source, target);
}
public void clear() {
playQueue.clear();
}
private Disposable getReactor() {
final Consumer<PlayQueueEvent> onNext = new Consumer<PlayQueueEvent>() {
@Override
@ -97,7 +94,7 @@ public class PlayQueueAdapter extends RecyclerView.Adapter<RecyclerView.ViewHold
}
};
return playQueue.getPlayQueueFlowable()
return playQueue.getEventBroadcast()
.toObservable()
.subscribe(onNext);
}

View File

@ -0,0 +1,19 @@
package org.schabi.newpipe.playlist.events;
public class AppendEvent implements PlayQueueMessage {
private int amount;
@Override
public PlayQueueEvent type() {
return PlayQueueEvent.APPEND;
}
public AppendEvent(final int amount) {
this.amount = amount;
}
public int getAmount() {
return amount;
}
}

View File

@ -0,0 +1,8 @@
package org.schabi.newpipe.playlist.events;
public class InitEvent implements PlayQueueMessage {
@Override
public PlayQueueEvent type() {
return PlayQueueEvent.INIT;
}
}

View File

@ -0,0 +1,19 @@
package org.schabi.newpipe.playlist.events;
public class NextEvent implements PlayQueueMessage {
private int newIndex;
@Override
public PlayQueueEvent type() {
return PlayQueueEvent.NEXT;
}
public NextEvent(final int newIndex) {
this.newIndex = newIndex;
}
public int index() {
return newIndex;
}
}

View File

@ -1,4 +1,4 @@
package org.schabi.newpipe.playlist;
package org.schabi.newpipe.playlist.events;
public enum PlayQueueEvent {
INIT,
@ -15,13 +15,7 @@ public enum PlayQueueEvent {
// sent when a pending stream is removed from the play queue
REMOVE,
// sent when the current stream is removed
REMOVE_CURRENT,
// sent when two streams swap place in the play queue
SWAP,
// sent when streams is cleared
CLEAR
SWAP
}

View File

@ -0,0 +1,5 @@
package org.schabi.newpipe.playlist.events;
public interface PlayQueueMessage {
PlayQueueEvent type();
}

View File

@ -0,0 +1,19 @@
package org.schabi.newpipe.playlist.events;
public class RemoveEvent extends PlayQueueMessage {
private int index;
@Override
public PlayQueueEvent type() {
return PlayQueueEvent.REMOVE;
}
public RemoveEvent(final int index) {
this.index = index;
}
public int index() {
return index;
}
}

View File

@ -0,0 +1,19 @@
package org.schabi.newpipe.playlist.events;
public class SelectEvent implements PlayQueueMessage {
private int newIndex;
@Override
public PlayQueueEvent type() {
return PlayQueueEvent.SELECT;
}
public SelectEvent(final int newIndex) {
this.newIndex = newIndex;
}
public int index() {
return newIndex;
}
}

View File

@ -0,0 +1,25 @@
package org.schabi.newpipe.playlist.events;
public class SwapEvent implements PlayQueueMessage {
private int from;
private int to;
@Override
public PlayQueueEvent type() {
return PlayQueueEvent.SWAP;
}
public SwapEvent(final int from, final int to) {
this.from = from;
this.to = to;
}
public int getFrom() {
return from;
}
public int getTo() {
return to;
}
}