为什么我的订阅从未完成

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);
                        }
                    }
                });
    }
}

然而，我的完全不跑步。我通过循环iApp的结果并输出其中一个字段来检查onNext，这样我就可以看到数据正在收集，但我的compareList函数从未运行过。

有人能解释一下我的疏忽吗？

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);
                        }
                    }
                });
    }