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

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

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 в workthread? Я вызываю функцию регистра, как показано ниже:

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 ()) - это успех, и советы Answers тоже успешны.

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