为什么我的订阅永远不会完成？

Question

我正在尝试学习 rxJava 和 android 上下文中的反应式编程，我觉得我快到了，我只是不能完全掌握完整的图片来完全理解我在做什么。

我有以下代码，它从数据库中获取名为 iApps 的类的实例列表

 myHelper m = new myHelper(getApplication());
        m.getApps()
                .observeOn(Schedulers.newThread())
                .subscribe(currentApps::addAll,
                        throwable -> Log.e("Error Observable", throwable.toString() + " " + Arrays.toString(throwable.getStackTrace())),
                        () -> compareLists(availableApps, currentApps));
}

其中使用以下方法： //来自我的数据库调用函数

public  Callable<ArrayList<iApp>> getApps()
    {
        return this::getCurrentInfo;
    }

自定义辅助函数

public class myHelper {

    Context ctx;
    tQuery t;
    public myHelper(Context _ctx)
    {
        this.ctx = _ctx;
        t = new tQuery(_ctx);
    }

    Observable<ArrayList<iApp>> getApps()
    {
        return makeObservable(t.getApps())
                .subscribeOn(Schedulers.computation());
    }

    private static <T> Observable<T> makeObservable(final Callable<T> func) {
        return Observable.create(
                new Observable.OnSubscribe<T>() {
                    @Override
                    public void call(Subscriber<? super T> subscriber) {
                        try {
                            subscriber.onNext(func.call());

                        } catch (Exception ex) {
                            subscriber.onError(ex);
                        }
                    }
                });
    }

}

但是我的 on complete 永远不会运行。我通过循环 iApp 的结果并输出其中一个字段来检查 onNext，以便我可以看到正在收集数据，但是我的 compareLists 函数从未运行。

谁能解释我的疏忽？

Answer 1

这太尴尬了！

private static <T> Observable<T> makeObservable(final Callable<T> func) {
        return Observable.create(
                new Observable.OnSubscribe<T>() {
                    @Override
                    public void call(Subscriber<? super T> subscriber) {
                        try {
                            subscriber.onNext(func.call());
                            subscriber.onCompleted();

                        } catch (Exception ex) {
                            subscriber.onError(ex);
                        }
                    }
                });
    }