Как обернуть слушателя с помощью rxjava2 и обработать переключатель потока? - PullRequest
Новая
       2

Как обернуть слушателя с помощью rxjava2 и обработать переключатель потока?

0 голосов
/ 28 августа 2018

Я обертываю слушателя с кодом ниже, 

public Observable<String> getIdToken() {
    return Observable.create(emitter -> {
        firebaseAuth.getAccessToken(false)
                .addOnSuccessListener(getTokenResult -> {
                    emitter.onNext(getTokenResult.getToken());
                })
                .addOnFailureListener(e -> {
                    if (e instanceof FirebaseAuthInvalidUserException) {
                        emitter.onError(new BaseException("", ERR_CODE_ACCOUNT_BANNED));
                    } else {
                        emitter.onError(e);
                    }
                });
    });
}

public Observable<BaseResponse> register() {
    return getIdToken()
            .flatMap(idToken -> userApi.register(idToken));
}

но встретить ошибку, 

.flatMap(idToken -> userApi.register(idToken));

эта подсказка кода работает в mainthread, я думаю 

firebaseAuth.getAccessToken(false)

внутренний код автоматически переключается в workthread и выполняет сетевой запрос к google firebase, при включении onSuccess внутренний код переключается в nowThread (MainTread), затем он делает 

            .subscribeOn(Schedulers.io())
            .observeOn(AndroidSchedulers.mainThread())

не работает. Но как обработать этот переключатель потока, чтобы сделать flatMap рабочим потоком? Я вызываю функцию register, как показано ниже: 

Disposable disposable = UserModel.getInstance().register()
            .subscribeOn(Schedulers.io())
            .observeOn(AndroidSchedulers.mainThread())
            .subscribe(baseResponse -> {
                Toast.makeText(this, "a", Toast.LENGTH_SHORT).show();
            }, throwable -> {
                throwable.printStackTrace();
                Log.d("MainActivity111", throwable.getMessage());
            });

Ошибка подсказки logcat: 

08-28 11:18:26.634 6870-6870/? W/System.err: android.os.NetworkOnMainThreadException
    at android.os.StrictMode$AndroidBlockGuardPolicy.onNetwork(StrictMode.java:1450)
    at java.net.AbstractPlainSocketImpl.doConnect(AbstractPlainSocketImpl.java:355)
    at java.net.AbstractPlainSocketImpl.connectToAddress(AbstractPlainSocketImpl.java:200)
    at java.net.AbstractPlainSocketImpl.connect(AbstractPlainSocketImpl.java:182)
    at java.net.SocksSocketImpl.connect(SocksSocketImpl.java:357)
    at java.net.Socket.connect(Socket.java:616)
    at okhttp3.internal.platform.AndroidPlatform.connectSocket(AndroidPlatform.java:71)
    at okhttp3.internal.connection.RealConnection.connectSocket(RealConnection.java:240)
    at okhttp3.internal.connection.RealConnection.connect(RealConnection.java:160)
    at okhttp3.internal.connection.StreamAllocation.findConnection(StreamAllocation.java:257)
    at okhttp3.internal.connection.StreamAllocation.findHealthyConnection(StreamAllocation.java:135)
    at okhttp3.internal.connection.StreamAllocation.newStream(StreamAllocation.java:114)
    at okhttp3.internal.connection.ConnectInterceptor.intercept(ConnectInterceptor.java:42)
    at okhttp3.internal.http.RealInterceptorChain.proceed(RealInterceptorChain.java:147)
    at okhttp3.internal.http.RealInterceptorChain.proceed(RealInterceptorChain.java:121)
    at okhttp3.internal.cache.CacheInterceptor.intercept(CacheInterceptor.java:93)
    at okhttp3.internal.http.RealInterceptorChain.proceed(RealInterceptorChain.java:147)

08-28 11:18:26.635 6870-6870/? W/System.err:     at okhttp3.internal.http.RealInterceptorChain.proceed(RealInterceptorChain.java:121)
    at okhttp3.internal.http.BridgeInterceptor.intercept(BridgeInterceptor.java:93)
    at okhttp3.internal.http.RealInterceptorChain.proceed(RealInterceptorChain.java:147)
08-28 11:18:26.636 6870-6870/? W/System.err:     at okhttp3.internal.http.RetryAndFollowUpInterceptor.intercept(RetryAndFollowUpInterceptor.java:126)
    at okhttp3.internal.http.RealInterceptorChain.proceed(RealInterceptorChain.java:147)
    at okhttp3.internal.http.RealInterceptorChain.proceed(RealInterceptorChain.java:121)
    at okhttp3.RealCall.getResponseWithInterceptorChain(RealCall.java:200)
    at okhttp3.RealCall.execute(RealCall.java:77)
08-28 11:18:26.637 6870-6870/? W/System.err:     at retrofit2.OkHttpCall.execute(OkHttpCall.java:180)
    at retrofit2.adapter.rxjava2.CallExecuteObservable.subscribeActual(CallExecuteObservable.java:42)
    at io.reactivex.Observable.subscribe(Observable.java:12036)
08-28 11:18:26.638 6870-6870/? W/System.err:     at retrofit2.adapter.rxjava2.BodyObservable.subscribeActual(BodyObservable.java:34)
    at io.reactivex.Observable.subscribe(Observable.java:12036)
    at io.reactivex.internal.operators.observable.ObservableFlatMap$MergeObserver.subscribeInner(ObservableFlatMap.java:165)
    at io.reactivex.internal.operators.observable.ObservableFlatMap$MergeObserver.onNext(ObservableFlatMap.java:139)
    at io.reactivex.internal.operators.observable.ObservableCreate$CreateEmitter.onNext(ObservableCreate.java:67)
08-28 11:18:26.639 6870-6870/? W/System.err:     at cn.candrwow.coincoin.model.UserModel.lambda$null$0(UserModel.java:37)
    at cn.candrwow.coincoin.model.-$$Lambda$UserModel$TP1OIyNide1WpDi-afDDpxaJhcE.onSuccess(Unknown Source:4)
    at com.google.android.gms.tasks.zzn.run(Unknown Source:27)
    at android.os.Handler.handleCallback(Handler.java:790)
08-28 11:18:26.640 6870-6870/? W/System.err:     at android.os.Handler.dispatchMessage(Handler.java:99)
    at android.os.Looper.loop(Looper.java:164)
    at android.app.ActivityThread.main(ActivityThread.java:6494)
    at java.lang.reflect.Method.invoke(Native Method)
    at com.android.internal.os.RuntimeInit$MethodAndArgsCaller.run(RuntimeInit.java:438)
    at com.android.internal.os.ZygoteInit.main(ZygoteInit.java:807)

Обновление

Я нашел поток переключателя внутреннего кода прослушивателя firebase на mainThread, а не поток, который его вызывает ，, так что проблема выше 

emitter.onNext(getTokenResult.getToken());

- это сила в mainThread, а не 

.subscribeOn(Schedulers.io())

внутренний код firebase не открыт, я регистрируюсь под кодом (это в Activity.onCreate) и нашел его принудительным переключателем: 

new Thread(() -> {
        Log.d("TestActivity", Thread.currentThread().getName());
        FirebaseAuth.getInstance().getAccessToken(false)
                .addOnSuccessListener(getTokenResult -> {
                    Log.d("TestActivity", Thread.currentThread().getName());
                });
    }).start();

logcat - это: 

    08-28 16:47:47.232 12289-12384/? D/TestActivity: Thread-5
main

Я добавляю .observeOn (Schedulers.io ()) - это успех, и совет Отвечающих - тоже успех. 

public Observable<BaseResponse> register() {
    return getIdToken()
//                .observeOn(Schedulers.io())
                .flatMap(idToken -> userApi.register(idToken).subscribeOn(Schedulers.io()));
    }

Это мой интерфейс UserApi: 

 public interface UserApi {
    @FormUrlEncoded
    @POST("account/register")
    Observable<BaseResponse> register(@Field("idToken") String idToken);

    @GET("account/testGoogle")
    Observable<BaseResponse> test(@Query("idToken") String idToken);
    }

1 Ответ

0 голосов
/ 28 августа 2018 
 .flatMap(idToken -> userApi.register(idToken))

Может быть 

 .flatMap(idToken -> userApi.register(idToken).subscribeOn(Schedulers.io()))

И тогда это будет работать.

...