Несбалансированный вход / выход на OkHttp

Так что у меня время от времени есть исключение, которое говорит java.lang.IllegalStateException: Unbalanced enter/exit, и это происходит, когда onStop моей деятельности вызывает disposables.clear().

Я спросил об этом раньше, и мне сказали, что это ошибка D8, но сейчас я использую последнюю версию D8 или что-то еще для Android Studio 3.1.3, и это все еще происходит.

Я также спросил об этом на github для OkHttp, и мне сказали найти способ вызвать call.cancel().

Итак, мой вопрос: как я могу заставить disposables.clear() вызывать call.cancel()?

Это исключение:

Caused by java.lang.IllegalStateException: Unbalanced enter/exit
    at okio.AsyncTimeout.enter(AsyncTimeout.java:73)
    at okio.AsyncTimeout$2.read(AsyncTimeout.java:235)
    at okio.RealBufferedSource.read(RealBufferedSource.java:47)
    at okhttp3.internal.http1.Http1Codec$AbstractSource.read(Http1Codec.java:363)
    at okhttp3.internal.http1.Http1Codec$ChunkedSource.read(Http1Codec.java:453)
    at okio.RealBufferedSource.read(RealBufferedSource.java:47)
    at okhttp3.internal.cache.CacheInterceptor$1.read(CacheInterceptor.java:174)
    at okhttp3.internal.Util.skipAll(Util.java:175)
    at okhttp3.internal.Util.discard(Util.java:157)
    at okhttp3.internal.cache.CacheInterceptor$1.close(CacheInterceptor.java:202)
    at okio.RealBufferedSource.close(RealBufferedSource.java:469)
    at okio.RealBufferedSource$1.close(RealBufferedSource.java:453)
    at java.io.BufferedInputStream.close(BufferedInputStream.java:472)
    at java.nio.channels.Channels$ReadableByteChannelImpl.implCloseChannel(Channels.java:255)
    at java.nio.channels.spi.AbstractInterruptibleChannel$1.interrupt(AbstractInterruptibleChannel.java:166)
    at java.lang.Thread.interrupt(Thread.java:956)
    at java.util.concurrent.FutureTask.cancel(FutureTask.java:146)
    at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.cancel(ScheduledThreadPoolExecutor.java:258)
    at io.reactivex.internal.schedulers.ScheduledRunnable.dispose(ScheduledRunnable.java:107)
    at io.reactivex.disposables.CompositeDisposable.dispose(CompositeDisposable.java:217)
    at io.reactivex.disposables.CompositeDisposable.dispose(CompositeDisposable.java:80)
    at io.reactivex.internal.schedulers.IoScheduler$EventLoopWorker.dispose(IoScheduler.java:210)
    at io.reactivex.Scheduler$DisposeTask.dispose(Scheduler.java:464)
    at io.reactivex.internal.disposables.DisposableHelper.dispose(DisposableHelper.java:125)
    at io.reactivex.internal.operators.observable.ObservableSubscribeOn$SubscribeOnObserver.dispose(ObservableSubscribeOn.java:74)
    at io.reactivex.internal.operators.observable.ObservableObserveOn$ObserveOnObserver.dispose(ObservableObserveOn.java:146)
    at io.reactivex.internal.disposables.DisposableHelper.dispose(DisposableHelper.java:125)
    at io.reactivex.observers.DisposableObserver.dispose(DisposableObserver.java:91)
    at io.reactivex.disposables.CompositeDisposable.dispose(CompositeDisposable.java:217)
    at io.reactivex.disposables.CompositeDisposable.clear(CompositeDisposable.java:183)
    at MyActivity.onStop(MyActivity.java:321)
    at android.app.Instrumentation.callActivityOnStop(Instrumentation.java:1289)
    at android.app.Activity.performStop(Activity.java:6841)
    at android.app.ActivityThread.performStopActivityInner(ActivityThread.java:3947)
    at android.app.ActivityThread.handleStopActivity(ActivityThread.java:4005)
    at android.app.ActivityThread.-wrap25(ActivityThread.java)
    at android.app.ActivityThread$H.handleMessage(ActivityThread.java:1561)
    at android.os.Handler.dispatchMessage(Handler.java:110)
    at android.os.Looper.loop(Looper.java:203)
    at android.app.ActivityThread.main(ActivityThread.java:6251)
    at java.lang.reflect.Method.invoke(Method.java)
    at com.android.internal.os.ZygoteInit$MethodAndArgsCaller.run(ZygoteInit.java:1063)
    at com.android.internal.os.ZygoteInit.main(ZygoteInit.java:924)

Код в основном такой:

Observable<Stuff> observable = Observable.create(new ObservableOnSubscribe<Stuff>() {
            @Override
            public void subscribe(@NonNull ObservableEmitter<Stuff> e) throws Exception {

           //do OkHttp stuff, call onNext(stuff). 

            }
        })
        .subscribeOn(Schedulers.io())
        .observeOn(AndroidSchedulers.mainThread());
        DisposableObserver<Stuff> disposableObserver = observable
            .subscribeWith(new DisposableObserver<Stuff>() {......});

        disposables.add(disposableObserver);
5
0
1 310
1

Ответы 1

e.setCancellable(new Cancellable() {
    @Override
    public void cancel() throws Exception {
        call.cancel();
    }
});

Вызовите setCancellable эмиттера внутри метода subscribe(@NonNull ObservableEmitter<Stuff> e), этот Cancellable будет выполнен при удалении Observable.

Мне любопытно, как вы вызываете метод close, в моем случае прерывание потока никогда не вызовет метод closeSource, мне нужно вызвать call.cancel() вручную, чтобы прервать блокирующее чтение.

Другие вопросы по теме