如何将我的异步任务转换为 Rxjava？

Question

我正在尝试将以下 2 个异步任务转换为 rxjava，但不确定如何进行。有任何想法吗？ :

 new AsyncTask<Void, Void, Void>() {
                @Override
                protected Void doInBackground(Void... params) {

                    /* Shutdown video players */
                    Set<Map.Entry<String, PlayerBundle>> entries = videoPlayerPool.entrySet();
                    for (Map.Entry<String, PlayerBundle> entry : entries) {
                        PlayerBundle bundle = entry.getValue();
                        bundle.player.release();
                    }

                    /* Shutdown audio players */
                    entries = audioPlayerPool.entrySet();
                    for (Map.Entry<String, PlayerBundle> entry : entries) {
                        PlayerBundle bundle = entry.getValue();
                        bundle.player.release();
                    }

                    videoStreamer.stopStream();
                    videoStreamer.release();

                    audioStreamer.stopStream();
                    audioStreamer.release();

                    return null;
                }

                @Override
                protected void onPostExecute(Void aVoid) {
                    cb.event(new Spin.Event<Void>());
                }
            }.execute();

和：

new AsyncTask<Void, Void, Void>() {
            @Override
            protected Void doInBackground(Void... params) {
                Set<Map.Entry<String, Participant>> entries = pool.entrySet();
                for (Map.Entry<String, Participant> entry : entries) {
                    Participant participant = entry.getValue();
                    participant.release();
                }
                return null;
            }

            @Override
            protected void onPostExecute(Void aVoid) {
                cb.event(new Spin.Event<Void>());
            }
        }.execute();

我已经在我的 gradle 中包含了 rxjava，但不确定如何转换它

Answer 1

    Observable.defer(new Func0<Observable<Void>>() {
        @Override
        public Observable<Void> call() {
            /* Shutdown video players */
            Set<Map.Entry<String, PlayerBundle>> entries = videoPlayerPool.entrySet();
            for (Map.Entry<String, PlayerBundle> entry : entries) {
                PlayerBundle bundle = entry.getValue();
                bundle.player.release();
            }

            /* Shutdown audio players */
            entries = audioPlayerPool.entrySet();
            for (Map.Entry<String, PlayerBundle> entry : entries) {
                PlayerBundle bundle = entry.getValue();
                bundle.player.release();
            }

            videoStreamer.stopStream();
            videoStreamer.release();

            audioStreamer.stopStream();
            audioStreamer.release();

            return Observable.just(null);
        }
    })
    .doOnCompleted(new Action0() {
        @Override
        public void call() {
            cb.event(new Spin.Event<Void>());
        }
    })
    .subscribeOn(Schedulers.computation())
    .subscribe();

    Observable.defer(new Func0<Observable<Void>>() {
        @Override
        public Observable<Void> call() {
            Set<Map.Entry<String, Participant>> entries = pool.entrySet();
            for (Map.Entry<String, Participant> entry : entries) {
                Participant participant = entry.getValue();
                participant.release();
            }
            return Observable.just(null);
        }
    }).doOnCompleted(new Action0() {
        @Override
        public void call() {
            cb.event(new Spin.Event<Void>());
        }
    })
    .subscribeOn(Schedulers.computation())
    .subscribe();