Rxjava: in this case subscribe onNext not working

[neodreamtale]

Thanks for using RxJava but before you post an issue, please consider the following points:

here is my retrofit and rxjava version：

//rx
compile 'io.reactivex.rxjava2:rxjava:2.0.3'
compile 'io.reactivex.rxjava2:rxandroid:2.0.1'
//retrofit
compile 'com.squareup.retrofit2:retrofit:2.2.0'
compile 'com.squareup.retrofit2:converter-gson:2.0.2'
compile 'com.squareup.retrofit2:adapter-rxjava2:2.2.0'
//okhttp3
compile 'com.squareup.okhttp3:okhttp:3.4.1'
compile 'com.squareup.okhttp3:logging-interceptor:3.4.1'
compile 'com.squareup.okio:okio:1.6.0'

retrofit like this
@GET("getAppList") Flowable<EBApiResult<List<UserAppInfo>>> testGetUserApp();

rxjava like this

Flowable<EBApiResult<List<UserAppInfo>>> flowable = testGetAppInfos();

    flowable.map(new Function<EBApiResult<List<UserAppInfo>>, List<UserAppInfo>>() {

        @Override
        public List<UserAppInfo> apply(EBApiResult<List<UserAppInfo>> result) throws Exception {
            return result.getResultData();
        }
    }).subscribeOn(Schedulers.io())
            .observeOn(AndroidSchedulers.mainThread())
            .subscribe(new Subscriber<List<UserAppInfo>>() {
                @Override
                public void onSubscribe(Subscription s) {
                    Log.i("123");
                }

                @Override
                public void onNext(List<UserAppInfo> userAppInfos) {
                    uiPresenter.bindAppInfoList(result);
                }

                @Override
                public void onError(Throwable t) {
                    uiPresenter.errPage(t.getMessage());
                }

                @Override
                public void onComplete() {
                    uiPresenter.errPage(null);
                }
            });`

I can see that onSubscribe is called back
but onNext onError is not。

If I call rxjava like this

Flowable<EBApiResult<List<UserAppInfo>>> flowable = testGetAppInfos();

  flowable.map(new Function<EBApiResult<List<UserAppInfo>>, List<UserAppInfo>>() {

        @Override
        public List<UserAppInfo> apply(EBApiResult<List<UserAppInfo>> result) throws Exception {
            return result.getResultData();
        }
    }).subscribeOn(Schedulers.io())
            .observeOn(AndroidSchedulers.mainThread())
            .subscribe(new Consumer<List<UserAppInfo>>() {
                @Override
                public void accept(List<UserAppInfo> appInfos) throws Exception {
                    storeAppInfos(appInfos);
                    List<UserAppInfo> result = filterListByCurrentCondition(appInfos, CURRENT_NEED_SHOW_APPS);
                    uiPresenter.bindAppInfoList(result);
                }
            }, new Consumer<Throwable>() {
                @Override
                public void accept(Throwable t) throws Exception {
                    uiPresenter.errPage(t.getMessage());
                }
            }, new Action() {
                @Override
                public void run() throws Exception {
                }
            });`

every accept can be reached
i dont know why。
but i debug with source，I doubt it because...my Flowable‘s implementation is not QueueSubscription。

I don't know how to fix my problem. I hope you can understand

[akarnokd]

You have to call s.request(Long.MAX_VALUE) in the first case where you have Log.i("123").

[neodreamtale]

@akarnokd thank you
Because of the language barrier, I didn't know how to use it before。After your prompt, I went to the relevant documents。it's working，thankyou！