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

package org.schabi.newpipe.playlist;

import android.util.Log;

import org.schabi.newpipe.extractor.InfoItem;
import org.schabi.newpipe.extractor.playlist.PlaylistInfo;
import org.schabi.newpipe.extractor.stream.StreamInfoItem;
import org.schabi.newpipe.util.ExtractorHelper;

import java.util.ArrayList;
import java.util.Collections;
import java.util.List;

import io.reactivex.SingleObserver;
import io.reactivex.android.schedulers.AndroidSchedulers;
import io.reactivex.annotations.NonNull;
import io.reactivex.disposables.Disposable;
import io.reactivex.schedulers.Schedulers;

public class ExternalPlayQueue extends PlayQueue {
    private final String TAG = "ExternalPlayQueue@" + Integer.toHexString(hashCode());
    private static final int RETRY_COUNT = 2;

    private boolean isComplete;

    private int serviceId;
    private String playlistUrl;

    private Disposable fetchReactor;

    public ExternalPlayQueue(final int serviceId,
                             final String nextPageUrl,
                             final List<InfoItem> streams,
                             final int index) {
        super(index, extractPlaylistItems(streams));

        this.playlistUrl = nextPageUrl;
        this.serviceId = serviceId;

        this.isComplete = nextPageUrl == null || nextPageUrl.isEmpty();
    }

    @Override
    public boolean isComplete() {
        return isComplete;
    }

    @Override
    public PlayQueueItem get(int index) {
        if (index > getStreams().size() || getStreams().get(index) == null) return null;
        return getStreams().get(index);
    }

    @Override
    public void fetch() {
       ExtractorHelper.getPlaylistInfo(this.serviceId, this.playlistUrl, false)
                .subscribeOn(Schedulers.io())
                .observeOn(AndroidSchedulers.mainThread())
                .retry(RETRY_COUNT)
                .subscribe(getPlaylistObserver());
    }

    private SingleObserver<PlaylistInfo> getPlaylistObserver() {
        return new SingleObserver<PlaylistInfo>() {
            @Override
            public void onSubscribe(@NonNull Disposable d) {
                if (isComplete || (fetchReactor != null && !fetchReactor.isDisposed())) {
                    d.dispose();
                } else {
                    fetchReactor = d;
                }
            }

            @Override
            public void onSuccess(@NonNull PlaylistInfo playlistInfo) {
                if (!playlistInfo.has_more_streams) isComplete = true;
                playlistUrl = playlistInfo.next_streams_url;

                append(extractPlaylistItems(playlistInfo.related_streams));
            }

            @Override
            public void onError(@NonNull Throwable e) {
                Log.e(TAG, "Error fetching more playlist, marking playlist as complete.", e);
                isComplete = true;
                append(Collections.<PlayQueueItem>emptyList());
            }
        };
    }

    @Override
    public void dispose() {
        super.dispose();
        if (fetchReactor != null) fetchReactor.dispose();
    }

    private static List<PlayQueueItem> extractPlaylistItems(final List<InfoItem> infos) {
        List<PlayQueueItem> result = new ArrayList<>();
        for (final InfoItem stream : infos) {
            if (stream instanceof StreamInfoItem) {
                result.add(new PlayQueueItem((StreamInfoItem) stream));
            }
        }
        return result;
    }
}