RxMediaPlayer.java

public class RxMediaPlayer {
    public static final int DEFAULT_SAMPLING_MILLIS = 33;
    public enum State {
        IDLE("idle"),
        INITIALIZED("initialize"),
        PREPARING("preparing"),
        PREPARED("prepared"),
        STARTED("started"),
        STOPPED("stopped"),
        PAUSED("paused"),
        PLAYBACK_COMPLETED("completed"),
        END("end"),
        ERROR("error");
        public final String name;
        State(String name) {
            this.name = name;
        }
    }

    private MediaPlayer mediaPlayer = new MediaPlayer();
    private Subscription samplingSubscription;
    private MediaPlayerListener listener = new MediaPlayerListener();
    private Scheduler scheduler = Schedulers.newThread();
    private String name = RxMediaPlayer.class.getSimpleName();
    private int samplingTimeMillis = DEFAULT_SAMPLING_MILLIS;

    final BehaviorSubject<Integer> durationSubject = BehaviorSubject.create(0);
    final BehaviorSubject<State> stateSubject = BehaviorSubject.create(State.IDLE);
    final BehaviorSubject<Integer> positionSubject = BehaviorSubject.create(0);
    final BehaviorSubject<Boolean> loopingSubject = BehaviorSubject.create(mediaPlayer.isLooping());
    final BehaviorSubject<Integer> bufferPercent = BehaviorSubject.create(0);
    final BehaviorSubject<Boolean> seekingSubject = BehaviorSubject.create(false);

    

    

    public RxMediaPlayer() {
        mediaPlayer.setOnErrorListener(listener);
        mediaPlayer.setOnCompletionListener(listener);
        mediaPlayer.setOnBufferingUpdateListener(listener);
        mediaPlayer.setOnSeekCompleteListener(listener);
        final Action1<State> onState = state -> {
            switch (state) {
                case STARTED:
                    subscribeSampling();
                    break;
                case PREPARED:
                case STOPPED:
                case PAUSED:
                case PLAYBACK_COMPLETED:
                case ERROR:
                    unsubscribeSampling();
                    break;
            }
        };
        getStateObs().subscribe(onState);
    }

    @NonNull
    public Observable<Integer> getPositionObs() {
        return positionSubject.asObservable();
    }
    public int getPosition() {
        return positionSubject.getValue();
    }

    @NonNull
    public Observable<State> getStateObs() {
        return stateSubject.asObservable();
    }
    public State getState() {
        return stateSubject.getValue();
    }


    public Scheduler myScheduler() {
        return scheduler;
    }

    public int getDuration() {
        return durationSubject.getValue();
    }
    @NonNull
    public Observable<Integer> getDurationObs() {
        return durationSubject.asObservable();
    }

    @NonNull
    public Observable<Integer> getBufferPercentObs() {
        return bufferPercent.asObservable();
    }

    @NonNull
    public Observable<Boolean> getSeekingObs() {
        return seekingSubject.asObservable();
    }

    public void setName(@NonNull String name) {
        this.name = name;
    }

    @NonNull
    public String getName() {
        return name;
    }

    public boolean isPlaying() {
        return mediaPlayer.isPlaying();
    }

    private void changeState(State state) {
        if (getState()!=state) {
            Timber.v(name + " changeState:" + state.name);
            stateSubject.onNext(state);
        }
    }

    private void setSeeking(boolean seeking) {
        if (seekingSubject.getValue()!=seeking) {
            //Timber.v("setSeeking:" + String.valueOf(seeking) + " " + name);
            seekingSubject.onNext(seeking);
        }
    }

    @NonNull
    public Observable<RxMediaPlayer> reset() {
        return Observable.create(new Observable.OnSubscribe<RxMediaPlayer>() {
            @Override
            public void call(Subscriber<? super RxMediaPlayer> subscriber) {
                if (subscriber.isUnsubscribed()) {
                    return;
                }

                try {
                    mediaPlayer.reset();
                    changeState(State.IDLE);
                    subscriber.onNext(getInstance());
                } catch (Exception e) {
                    subscriber.onError(e);
                }
            }
        });
    }

    @NonNull
    public Observable<RxMediaPlayer> setDataSource(@NonNull String path) {
        return Observable.create(new Observable.OnSubscribe<RxMediaPlayer>() {
            @Override
            public void call(Subscriber<? super RxMediaPlayer> subscriber) {
                if (subscriber.isUnsubscribed()) {
                    return;
                }

                if (!canSetDataSource()) {
                    final String message = composeIllegalStateMessage("setDataSource");
                    subscriber.onError(new IllegalStateException(message));
                } else {
                    try {
                        mediaPlayer.setDataSource(path);
                        changeState(State.INITIALIZED);
                        subscriber.onNext(getInstance());
                    } catch (IOException e) {
                        subscriber.onError(e);
                    }
                }
            }
        });
    }

    @NonNull
    public Observable<RxMediaPlayer> prepare() {
        return Observable.create(new Observable.OnSubscribe<RxMediaPlayer>() {
            @Override
            public void call(Subscriber<? super RxMediaPlayer> subscriber) {
                if (subscriber.isUnsubscribed()) {
                    return;
                }

                if (!canPrepare()) {
                    final String message = composeIllegalStateMessage("prepare");
                    subscriber.onError(new IllegalStateException(message));
                } else {
                    try {
                        changeState(State.PREPARING);
                        mediaPlayer.prepare();
                        durationSubject.onNext(mediaPlayer.getDuration());
                        changeState(State.PREPARED);
                        subscriber.onNext(getInstance());
                    } catch (IOException e) {
                        subscriber.onError(e);
                    }
                }
            }
        });
    }

    @NonNull
    public Observable<RxMediaPlayer> start() {
        return Observable.create(new Observable.OnSubscribe<RxMediaPlayer>() {
            @Override
            public void call(Subscriber<? super RxMediaPlayer> subscriber) {
                if (subscriber.isUnsubscribed()) {
                    return;
                }

                if (!canStart()) {
                    final String message = composeIllegalStateMessage("start");
                    subscriber.onError(new IllegalStateException(message));
                } else {
                    try {
                        mediaPlayer.start();
                        changeState(State.STARTED);
                        subscriber.onNext(getInstance());
                    } catch (Exception e) {
                        subscriber.onError(e);
                    }
                }
            }
        });
    }

    @NonNull
    public Observable<RxMediaPlayer> seekTo(int position) {
        //Timber.v("seekTo..." + name);
        if (seekingSubject.getValue()) {
            Timber.w("seeking is true???");
            return Observable.just(this);
        }

        return Observable.create(new Observable.OnSubscribe<RxMediaPlayer>() {
            @Override
            public void call(Subscriber<? super RxMediaPlayer> subscriber) {
                if (subscriber.isUnsubscribed()) {
                    return;
                }

                if (!canSeek()) {
                    final String message = composeIllegalStateMessage("seekTo");
                    subscriber.onError(new IllegalStateException(message));
                } else {
                    try {
                        mediaPlayer.seekTo(position);
                        setSeeking(true);
                        subscriber.onNext(getInstance());
                    } catch (Exception e) {
                        subscriber.onError(e);
                    }
                }
            }
        });
    }

    @NonNull
    public Observable<RxMediaPlayer> pause() {
        return Observable.create(new Observable.OnSubscribe<RxMediaPlayer>() {
            @Override
            public void call(Subscriber<? super RxMediaPlayer> subscriber) {
                if (subscriber.isUnsubscribed()) {
                    return;
                }

                if (!canPause()) {
                    final String message = composeIllegalStateMessage("pause");
                    subscriber.onError(new IllegalStateException(message));
                } else {
                    try {
                        mediaPlayer.pause();
                        changeState(State.PAUSED);
                        subscriber.onNext(getInstance());
                    } catch (Exception e) {
                        subscriber.onError(e);
                    }
                }
            }
        });
    }

    @NonNull
    public Observable<RxMediaPlayer> stop() {
        return Observable.create(new Observable.OnSubscribe<RxMediaPlayer>() {
            @Override
            public void call(Subscriber<? super RxMediaPlayer> subscriber) {
                if (subscriber.isUnsubscribed()) {
                    return;
                }

                if (!canStop()) {
                    final String message = composeIllegalStateMessage("stop");
                    subscriber.onError(new IllegalStateException(message));
                } else {
                    try {
                        mediaPlayer.stop();
                        changeState(State.STOPPED);
                        subscriber.onNext(getInstance());
                    } catch (Exception e) {
                        subscriber.onError(e);
                    }
                }
            }
        });
    }

    public void release() {
        unsubscribeSampling();
        mediaPlayer.release();
        changeState(State.END);
    }

    public void setDisplay(SurfaceHolder sh) {
        mediaPlayer.setDisplay(sh);
    }

    public void setLooping(boolean looping) {
        if (looping!=mediaPlayer.isLooping()) {
            mediaPlayer.setLooping(looping);
            loopingSubject.onNext(mediaPlayer.isLooping());
        }
    }

    public boolean isLooping() {
        return loopingSubject.getValue();
    }

    public Observable<Boolean> getLoopingObs() {
        return loopingSubject.asObservable();
    }

    String composeIllegalStateMessage(String op) {
        StringBuilder sb = new StringBuilder();
        sb.append("[").append(name).append("] ");
        sb.append("Can't \'").append(op).append("\'. Current state:");
        sb.append(getState().name);
        return sb.toString();
    }

    RxMediaPlayer getInstance() {
        return this;
    }

    public boolean canSetDataSource() {
        switch (getState()) {
            case IDLE:
                return true;
            default:
                return false;
        }
    }

    public boolean canPause() {
        switch (getState()) {
            case STARTED:
            case PAUSED:
                return true;
            default:
                return false;
        }
    }

    public boolean canStop() {
        switch (getState()) {
            case STARTED:
            case STOPPED:
            case PAUSED:
            case PREPARED:
            case PLAYBACK_COMPLETED:
                return true;
            default:
                return false;
        }
    }

    boolean canStart() {
        switch (getState()) {
            case PREPARED:
            case STARTED:
            case PAUSED:
            case PLAYBACK_COMPLETED:
                return true;
            default:
                return false;
        }
    }

    boolean canPrepare() {
        switch (getState()) {
            case INITIALIZED:
            case STOPPED:
                return true;
            default:
                return false;
        }
    }

    public boolean canSeek() {
        switch (getState()) {
            case PREPARED:
            case STARTED:
            case PAUSED:
            case PLAYBACK_COMPLETED:
                return true;
            default:
                return false;
        }
    }

    void subscribeSampling() {
        unsubscribeSampling();
        samplingSubscription = Observable.timer(samplingTimeMillis, TimeUnit.MILLISECONDS)
                .subscribe(this::onSamplePosition, RxUtils.slientError());
    }

    void unsubscribeSampling() {
        if (samplingSubscription !=null) {
            samplingSubscription.unsubscribe();
            samplingSubscription = null;
        }
    }

    private int previousPosition = -1;
    void onSamplePosition(Long t) {
        int currentPosition = mediaPlayer.getCurrentPosition();
        if (currentPosition > previousPosition) {
            setSeeking(false);
        }
        positionSubject.onNext(currentPosition);
        previousPosition = currentPosition;
        subscribeSampling();
    }

    @Override
    public String toString() {
        return "name";
    }

    class MediaPlayerListener implements MediaPlayer.OnCompletionListener,
            MediaPlayer.OnErrorListener,
            MediaPlayer.OnBufferingUpdateListener,
            MediaPlayer.OnSeekCompleteListener,
            MediaPlayer.OnInfoListener
    {

        @Override
        public void onCompletion(MediaPlayer mp) {
            //Timber.v("onCompletion");
            if (getState()!=State.PLAYBACK_COMPLETED) {
                changeState(State.PLAYBACK_COMPLETED);
            }

        }

        @Override
        public boolean onError(MediaPlayer mp, int what, int extra) {
            Timber.e(name + ".onError, What:" + String.valueOf(what));
            changeState(State.ERROR);
            return false;
        }

        @Override
        public void onBufferingUpdate(MediaPlayer mediaPlayer, int percent) {
            bufferPercent.onNext(percent);
            //Timber.v("onBufferingUpdate:%d percent", percent);
        }

        @Override
        public void onSeekComplete(MediaPlayer mediaPlayer) {
            //Timber.v("onSeekComplete:" + name);
            setSeeking(false);
        }

        @Override
        public boolean onInfo(MediaPlayer mediaPlayer, int what, int extra) {
            Timber.v(name + ".onInfo, What:" + String.valueOf(what) + " Extra:" + String.valueOf(extra));
            return false;
        }
    }
}