RxJava 从 Observable 更新数据

Question

我有 2 个列表视图。它包含文本项。我想将文本项从左侧列表拖放到右侧列表。对此我没有问题。另外，我有一个 Observable，它发出 KalturaVideo 对象。这是我如何获得此可观察对象的代码：此函数获取已添加到右侧列表视图中的 id，并从结果中排除对象，如果对象 id 等于 targetList 中的某个 id:

private Observable<KalturaVideo> getSourceListObservalbe(String kalturaPlaylistId){
        Observable<String[]> targetIdList = Observable.just(targetList)
                .map((List<KalturaVideo> kalturaVideos) -> {
                    if(kalturaVideos == null || kalturaVideos.isEmpty()){
                        return new String[]{""};
                    }
                    String[] result = new String[kalturaVideos.size()];
                    int index = 0;
                    for (KalturaVideo item : kalturaVideos) {
                        result[index] = item.getId();
                    }
                    return result;
                });
        return Observable.combineLatest(
                targetIdList,
                KalturaVideoRetriver.getVideoList(BuildPlaylistStep2Activity.this, kalturaPlaylistId),
                (String[] idListOfTarget, KalturaVideo kalturaVideo) -> {
                    for (String item :idListOfTarget){
                        if(item.equals(kalturaVideo.getId())){
                            return null;
                        }
                    }
                    return kalturaVideo;
                })
                .doOnError(throwable -> {
                    int i =0;
                    i++;
                   throwable.printStackTrace();
                })
                .filter(kalturaVideo -> {
                    return kalturaVideo != null;
                });
    }

这里是视频检索器的代码：

public class KalturaVideoRetriver {

    public static final String KALTURA_NEW_PLAYLIST_ID = "NEW_PLAYLIST";

    public static Observable<KalturaVideo> getVideoList(Context context, String kalturaPlaylistId){
        return Observable.create(new Observable.OnSubscribe<KalturaVideo>() {
            @Override
            public void call(Subscriber<? super KalturaVideo> subscriber) {
                getKalturaPlaylistContentObservable(kalturaPlaylistId, context)
                        .flatMap(new Func1<String[], Observable<?>>() {
                            @Override
                            public Observable<?> call(String[] videoIdList) {
                                return getKalturaVideoListObservable(context, videoIdList)
                                        .doOnNext(kalturaVideo -> {
                                            subscriber.onNext(kalturaVideo);
                                        })
                                        .doOnCompleted(() -> subscriber.onCompleted());
                            }
                        })
                        .subscribe();
            }
        });
    }

    private static Observable<KalturaVideo> getKalturaVideoListObservable(Context context, String[] kalturaVideoIdArray){
        return Observable.create(new Observable.OnSubscribe<KalturaVideo>() {
            @Override
            public void call(Subscriber<? super KalturaVideo> subscriber) {
                Cursor query;
                List<KalturaVideo> result = new ArrayList<>();
                if(kalturaVideoIdArray == null || kalturaVideoIdArray.length == 0) {
                    query = context.getContentResolver().query(KalturaVideoColumns.CONTENT_URI, null, null, null, null);
                }else {
                    KalturaVideoSelection where = new KalturaVideoSelection();
                    where.kalturaIdLike(kalturaVideoIdArray);
                    query = context.getContentResolver().query(
                            KalturaVideoColumns.CONTENT_URI,
                            null,
                            where.sel(),
                            where.args(),
                            null);
                }
                if(!query.moveToFirst()){
                    query.close();
                    subscriber.onCompleted();
                }
                KalturaVideoCursor cursor = new KalturaVideoCursor(query);
                do{
                    KalturaVideo video = new KalturaVideo();
                    video.setId(cursor.getKalturaId());
                    video.setName(cursor.getName());
                    video.setDescription(cursor.getDescription());
                    video.setCategories(cursor.getCategories());
                    video.setCategoriesIds(cursor.getCategoriesIds());
                    video.setDownloadUrl(cursor.getDownloadUrl());
                    video.setThumbnailUrl(cursor.getThumbnailUrl());
                    video.setDataUrl(cursor.getDataUrl());
                    video.setDuration(cursor.getDuration());
                    subscriber.onNext(video);
                }while (cursor.moveToNext());
                query.close();
                subscriber.onCompleted();
            }
        });
    }

    private static Observable<String[]> getKalturaPlaylistContentObservable(String kalturaPlaylistId, Context context){
        return Observable.create(new Observable.OnSubscribe<String[]>() {
            @Override
            public void call(Subscriber<? super String[]> subscriber) {
                if(kalturaPlaylistId.equals(KALTURA_NEW_PLAYLIST_ID)){
                    subscriber.onNext(new String[]{});
                    subscriber.onCompleted();
                    return;
                }
                KalturaPlaylistContentSelection where = new KalturaPlaylistContentSelection();
                where.playlistId(kalturaPlaylistId);
                Cursor query = context.getContentResolver().query(
                        KalturaPlaylistContentColumns.CONTENT_URI,
                        null,
                        where.sel(),
                        where.args(),
                        null
                );
                if(!query.moveToFirst()){
                    subscriber.onNext(new String[]{});
                    query.close();
                    subscriber.onCompleted();
                    return;
                }
                KalturaPlaylistContentCursor cursor = new KalturaPlaylistContentCursor(query);
                String[] result = new String[query.getCount()];
                int index = 0;
                do{
                    result[index] = cursor.getKalturaVideoId();
                    index++;
                }while (query.moveToNext());
                query.close();
                subscriber.onNext(result);
                subscriber.onCompleted();
            }
        });

    }

}

首先，在检索器中，我通过播放列表 ID 获取视频 ID。如果 PlaylistId 为空 - 我会得到所有存储在我的数据库中的视频 ID。在此之后，我通过 id 从 db 获取所有视频，这是我在上一步中获得的。

这是一个函数，可以帮助我获取所有视频并显示它：

private Observable<KalturaVideo> fillLists(){
        return getSourceListObservalbe(KalturaVideoRetriver.KALTURA_NEW_PLAYLIST_ID)
                .subscribeOn(Schedulers.io())
                .observeOn(AndroidSchedulers.mainThread())
                .doOnError(throwable -> {
                    Log.e("TAG", throwable.getCause().toString());
                })
                .doOnNext(kalturaVideo -> {
                    sourceList.add(kalturaVideo);
                })
                .doOnCompleted(() -> {
                    viewHolder.sourceListView.setAdapter(new BuildPlaylistContentAdapter(
                            BuildPlaylistStep2Activity.this,
                            -1,
                            sourceList));
                    viewHolder.sourceListView.setOnItemLongClickListener(new AdapterView.OnItemLongClickListener() {
                        @Override
                        public boolean onItemLongClick(AdapterView<?> parent, View view, int position, long id) {
                            KalturaVideo video = (KalturaVideo) view.getTag();
                            ClipData data = ClipData.newPlainText("kalturaVideoId", video.getId());
                            View.DragShadowBuilder shadowBuilder = new View.DragShadowBuilder(view);
                            view.startDrag(data, shadowBuilder, view, 0);
                            view.setVisibility(View.INVISIBLE);
                            return false;
                        }
                    });
                    viewHolder.targetListView.setAdapter(new BuildPlaylistContentAdapter(
                            BuildPlaylistStep2Activity.this,
                            -1,
                            targetList
                    ));
                });
    }

我是这样写的：

subscription = fillLists().subscribe();

下面是我如何尝试在拖放后更新 2 个列表：

targetList.add(video);
sourceList.clear();
fillLists().subscribe();

其中视频 - 是与视图相关的视频，已被删除到新列表。

因此，当我第一次将视图拖放到另一个列表时，效果很好。但是当我第二次下降时，我在 doOnError 方法中得到了这个堆栈跟踪：

04-20 18:46:02.072 8015-8083/com.mediamanagment.app W/System.err﹕ java.lang.NullPointerException 04-20 18:46:02.072
8015-8083/com.mediamanagment.app W/System.err﹕ at com.mediamanagment.app.activity.BuildPlaylistStep2Activity.lambda$getSourceListObservalbe(BuildPlaylistStep2Activity.java:106) 04-20 18:46:02.072 8015-8083/com.mediamanagment.app W/System.err﹕ at com.mediamanagment.app.activity.BuildPlaylistStep2Activity.access$lambda(BuildPlaylistStep2Activity.java) 04-20 18:46:02.072 8015-8083/com.mediamanagment.app W/System.err﹕ at com.mediamanagment.app.activity.BuildPlaylistStep2Activity$$Lambda.call(Unknown Source) 04-20 18:46:02.080 8015-8083/com.mediamanagment.app W/System.err﹕ at rx.functions.Functions.call(Functions.java:76) 04-20 18:46:02.080 8015-8083/com.mediamanagment.app W/System.err﹕ at rx.operators.OperatorCombineLatest$Collector.next(OperatorCombineLatest.java:157) 04-20 18:46:02.080 8015-8083/com.mediamanagment.app W/System.err﹕ at rx.operators.OperatorCombineLatest$SourceSubscriber.onNext(OperatorCombineLatest.java:218) 04-20 18:46:02.080 8015-8083/com.mediamanagment.app W/System.err﹕ at com.mediamanagment.app.retrivers.KalturaVideoRetriver.lambda$call(KalturaVideoRetriver.java:36) 04-20 18:46:02.080 8015-8083/com.mediamanagment.app W/System.err﹕ at com.mediamanagment.app.retrivers.KalturaVideoRetriver.access$lambda[=15=](KalturaVideoRetriver.java) 04-20 18:46:02.080 8015-8083/com.mediamanagment.app W/System.err﹕ at com.mediamanagment.app.retrivers.KalturaVideoRetriver$$Lambda.call(Unknown Source) 04-20 18:46:02.080 8015-8083/com.mediamanagment.app W/System.err﹕ at rx.Observable.onNext(Observable.java:3682) 04-20 18:46:02.080 8015-8083/com.mediamanagment.app W/System.err﹕ at rx.operators.OperatorDoOnEach.onNext(OperatorDoOnEach.java:61) 04-20 18:46:02.080 8015-8083/com.mediamanagment.app W/System.err﹕ at com.mediamanagment.app.retrivers.KalturaVideoRetriver.call(KalturaVideoRetriver.java:80) 04-20 18:46:02.080 8015-8083/com.mediamanagment.app W/System.err﹕ at com.mediamanagment.app.retrivers.KalturaVideoRetriver.call(KalturaVideoRetriver.java:47) 04-20 18:46:02.080 8015-8083/com.mediamanagment.app W/System.err﹕ at rx.Observable.call(Observable.java:153) 04-20 18:46:02.080
8015-8083/com.mediamanagment.app W/System.err﹕ at rx.Observable.call(Observable.java:149) 04-20 18:46:02.080
8015-8083/com.mediamanagment.app W/System.err﹕ at rx.Observable.call(Observable.java:153) 04-20 18:46:02.080
8015-8083/com.mediamanagment.app W/System.err﹕ at rx.Observable.call(Observable.java:149) 04-20 18:46:02.080
8015-8083/com.mediamanagment.app W/System.err﹕ at rx.Observable.unsafeSubscribe(Observable.java:6110) 04-20 18:46:02.080 8015-8083/com.mediamanagment.app W/System.err﹕ at rx.operators.OperatorMerge.onNext(OperatorMerge.java:66) 04-20 18:46:02.080 8015-8083/com.mediamanagment.app W/System.err﹕ at rx.operators.OperatorMerge.onNext(OperatorMerge.java:43) 04-20 18:46:02.080 8015-8083/com.mediamanagment.app W/System.err﹕ at rx.operators.OperatorMap.onNext(OperatorMap.java:54) 04-20 18:46:02.080 8015-8083/com.mediamanagment.app W/System.err﹕ at com.mediamanagment.app.retrivers.KalturaVideoRetriver.call(KalturaVideoRetriver.java:93) 04-20 18:46:02.080 8015-8083/com.mediamanagment.app W/System.err﹕ at com.mediamanagment.app.retrivers.KalturaVideoRetriver.call(KalturaVideoRetriver.java:89) 04-20 18:46:02.080 8015-8083/com.mediamanagment.app W/System.err﹕ at rx.Observable.call(Observable.java:153) 04-20 18:46:02.080
8015-8083/com.mediamanagment.app W/System.err﹕ at rx.Observable.call(Observable.java:149) 04-20 18:46:02.080
8015-8083/com.mediamanagment.app W/System.err﹕ at rx.Observable.call(Observable.java:153) 04-20 18:46:02.088
8015-8083/com.mediamanagment.app W/System.err﹕ at rx.Observable.call(Observable.java:149) 04-20 18:46:02.088
8015-8083/com.mediamanagment.app W/System.err﹕ at rx.Observable.subscribe(Observable.java:6177) 04-20 18:46:02.088
8015-8083/com.mediamanagment.app W/System.err﹕ at rx.Observable.subscribe(Observable.java:5842) 04-20 18:46:02.088
8015-8083/com.mediamanagment.app W/System.err﹕ at com.mediamanagment.app.retrivers.KalturaVideoRetriver.call(KalturaVideoRetriver.java:41) 04-20 18:46:02.088 8015-8083/com.mediamanagment.app W/System.err﹕ at com.mediamanagment.app.retrivers.KalturaVideoRetriver.call(KalturaVideoRetriver.java:27) 04-20 18:46:02.088 8015-8083/com.mediamanagment.app W/System.err﹕ at rx.Observable.unsafeSubscribe(Observable.java:6110) 04-20 18:46:02.095 8015-8083/com.mediamanagment.app W/System.err﹕ at rx.operators.OperatorCombineLatest.call(OperatorCombineLatest.java:87) 04-20 18:46:02.095 8015-8083/com.mediamanagment.app W/System.err﹕ at rx.operators.OperatorCombineLatest.call(OperatorCombineLatest.java:40) 04-20 18:46:02.095 8015-8083/com.mediamanagment.app W/System.err﹕ at rx.Observable.call(Observable.java:153) 04-20 18:46:02.095
8015-8083/com.mediamanagment.app W/System.err﹕ at rx.Observable.call(Observable.java:149) 04-20 18:46:02.095
8015-8083/com.mediamanagment.app W/System.err﹕ at rx.Observable.call(Observable.java:153) 04-20 18:46:02.095
8015-8083/com.mediamanagment.app W/System.err﹕ at rx.Observable.call(Observable.java:149) 04-20 18:46:02.095
8015-8083/com.mediamanagment.app W/System.err﹕ at rx.Observable.unsafeSubscribe(Observable.java:6110) 04-20 18:46:02.095 8015-8083/com.mediamanagment.app W/System.err﹕ at rx.operators.OperatorSubscribeOn.call(OperatorSubscribeOn.java:60) 04-20 18:46:02.095 8015-8083/com.mediamanagment.app W/System.err﹕ at rx.schedulers.NewThreadScheduler$NewThreadWorker$ScheduledAction.run(NewThreadScheduler.java:141) 04-20 18:46:02.095 8015-8083/com.mediamanagment.app W/System.err﹕ at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:390) 04-20 18:46:02.095 8015-8083/com.mediamanagment.app W/System.err﹕ at java.util.concurrent.FutureTask.run(FutureTask.java:234) 04-20 18:46:02.095 8015-8083/com.mediamanagment.app W/System.err﹕ at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.access1(ScheduledThreadPoolExecutor.java:153) 04-20 18:46:02.103 8015-8083/com.mediamanagment.app W/System.err﹕ at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:267) 04-20 18:46:02.103 8015-8083/com.mediamanagment.app W/System.err﹕ at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1080) 04-20 18:46:02.103 8015-8083/com.mediamanagment.app W/System.err﹕ at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:573) 04-20 18:46:02.103 8015-8083/com.mediamanagment.app W/System.err﹕ at java.lang.Thread.run(Thread.java:856)

没有返回任何视频

Answer 1

您在此代码中遇到了 NullPointerException

 Log.e("TAG", throwable.getCause().toString());

我认为您的异常没有原因，因此 toString 调用抛出 NPE (NullPointerException)

将您的代码替换为

Log.e("TAG", throwable.getCause());

应该可以解决这个问题

但是您的代码非常复杂，有些部分可以用不同的方式编写。还有一些是越野车。

例如：

错误

.doOnError(throwable -> {
    int i =0;
    i++;
    throwable.printStackTrace();
})

i 将始终等于 0（或在 i++ 之后等于 1）。是自愿的吗？

错误代码

public static Observable<KalturaVideo> getVideoList(Context context, String kalturaPlaylistId){
    return Observable.create(new Observable.OnSubscribe<KalturaVideo>()     {
        @Override
        public void call(Subscriber<? super KalturaVideo> subscriber) {
            getKalturaPlaylistContentObservable(kalturaPlaylistId, context)
                    .flatMap(new Func1<String[], Observable<?>>() {
                        @Override
                        public Observable<?> call(String[] videoIdList) {
                            return getKalturaVideoListObservable(context, videoIdList)
                                    .doOnNext(kalturaVideo -> {
                                        subscriber.onNext(kalturaVideo);
                                    })
                                    .doOnCompleted(() -> subscriber.onCompleted());
                        }
                    })
                    .subscribe();
        }
    });
}

不要在 Observable 中使用 Observable 。只改造第一个

public static Observable<KalturaVideo> getVideoList(Context context, String kalturaPlaylistId){ return getKalturaPlaylistContentObservable(kalturaPlaylistId, context).flatMap((String[] videoIdList) -> getKalturaVideoListObservable(context, videoIdList)); }

复码

Observable<String[]> targetIdList = Observable.just(targetList) .map((List<KalturaVideo> kalturaVideos) -> { if(kalturaVideos == null || kalturaVideos.isEmpty()){ return new String[]{""}; } String[] result = new String[kalturaVideos.size()]; int index = 0; for (KalturaVideo item : kalturaVideos) { result[index] = item.getId(); } return result; });

您尝试重写 Rx 本身已经可以完成的代码。

Observable<String[]> targetIdList = Observable.just(targetList) .flatMapIterable(videos -> videos) .map(video -> video.getId()) .toList() .map(l -> l.toArray(new String[]));

RxJava 从 Observable 更新数据

RxJava update data from Observable

java

reactive-programming

rx-java