{"action":"closed","issue":{"url":"https://api.github.com/repos/ReactiveX/RxJava/issues/3024","labels_url":"https://api.github.com/repos/ReactiveX/RxJava/issues/3024/labels{/name}","comments_url":"https://api.github.com/repos/ReactiveX/RxJava/issues/3024/comments","events_url":"https://api.github.com/repos/ReactiveX/RxJava/issues/3024/events","html_url":"https://github.com/ReactiveX/RxJava/issues/3024","id":89078546,"number":3024,"title":"Race condition between onNext and unsubscribe","user":{"login":"daschl","id":29086,"avatar_url":"https://avatars.githubusercontent.com/u/29086?v=3","gravatar_id":"","url":"https://api.github.com/users/daschl","html_url":"https://github.com/daschl","followers_url":"https://api.github.com/users/daschl/followers","following_url":"https://api.github.com/users/daschl/following{/other_user}","gists_url":"https://api.github.com/users/daschl/gists{/gist_id}","starred_url":"https://api.github.com/users/daschl/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/daschl/subscriptions","organizations_url":"https://api.github.com/users/daschl/orgs","repos_url":"https://api.github.com/users/daschl/repos","events_url":"https://api.github.com/users/daschl/events{/privacy}","received_events_url":"https://api.github.com/users/daschl/received_events","type":"User","site_admin":false},"labels":[{"url":"https://api.github.com/repos/ReactiveX/RxJava/labels/Question","name":"Question","color":"cc317c"}],"state":"closed","locked":false,"assignee":null,"milestone":null,"comments":32,"created_at":"2015-06-17T17:37:02Z","updated_at":"2015-09-03T07:54:33Z","closed_at":"2015-09-03T07:54:33Z","body":"Hi folks,\r\n\r\nI've seen some threads on this threaded message, but I'm still getting it on 1.10 - 1.12. Maybe it's something in my code, but I'm not sure what's going on. I have couchbase-related code which looks like:\r\n\r\n```java\r\nbucket\r\n    .async()\r\n    .getFromReplica(\"id::0\", ReplicaMode.ALL)\r\n    .toBlocking()\r\n    .first();\r\n```\r\n\r\nThe exact semantics don't matter, the important part is that the `getFromReplica` could return N responses, but we take only the first, meaning we eagerly unsubscribe. Now I need to do buffer freeing and so forth, so @akarnokd recommended me a lift operator. It works fine, but the eager unsubscribing fails. The code in question is:\r\n\r\n```java\r\npublic static <T> Observable<T> liftForAutoRelease(final Observable<T> source) {\r\n    return source.lift(new Observable.Operator<T, T>() {\r\n        @Override\r\n        public Subscriber<? super T> call(final Subscriber<? super T> child) {\r\n            return new Subscriber<T>(child) {\r\n                @Override\r\n                public void onCompleted() {\r\n                    child.onCompleted();\r\n                }\r\n\r\n                @Override\r\n                public void onError(Throwable e) {\r\n                    child.onError(e);\r\n                }\r\n\r\n                @Override\r\n                public void onNext(T t) {\r\n                    if (!child.isUnsubscribed()) {\r\n                        child.onNext(t);\r\n                    } else {\r\n                        ReferenceCountUtil.release(t);\r\n                    }\r\n                }\r\n            };\r\n        }\r\n    });\r\n}\r\n```\r\n\r\nAnd the `source` is an `AsyncSubject` made cold:\r\n\r\n```java\r\n@Override\r\npublic <R extends CouchbaseResponse> Observable<R> send (final RequestFactory requestFactory) {\r\n    return Buffers.liftForAutoRelease(Observable.defer(new Func0<Observable<R>>() {\r\n        @Override\r\n        public Observable<R> call() {\r\n            return sendHot(requestFactory.call());\r\n        }\r\n    }));\r\n}\r\n```\r\n\r\nNow the dreaded error is:\r\n\r\n```\r\njava.lang.IllegalStateException: This instance has been unsubscribed and the queue is no longer usable.\r\n\tat rx.internal.util.RxRingBuffer.onNext(RxRingBuffer.java:346)\r\n\tat rx.internal.operators.OperatorMerge$InnerSubscriber.enqueue(OperatorMerge.java:721)\r\n\tat rx.internal.operators.OperatorMerge$InnerSubscriber.emit(OperatorMerge.java:698)\r\n\tat rx.internal.operators.OperatorMerge$InnerSubscriber.onNext(OperatorMerge.java:586)\r\n\tat com.couchbase.client.core.utils.Buffers$2$1.onNext(Buffers.java:86)\r\n\tat rx.subjects.SubjectSubscriptionManager$SubjectObserver.onNext(SubjectSubscriptionManager.java:224)\r\n\tat rx.subjects.AsyncSubject.onCompleted(AsyncSubject.java:101)\r\n\tat com.couchbase.client.core.endpoint.AbstractGenericHandler$1.call(AbstractGenericHandler.java:208)\r\n\tat rx.internal.schedulers.ScheduledAction.run(ScheduledAction.java:55)\r\n\tat java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)\r\n\tat java.util.concurrent.FutureTask.run(FutureTask.java:266)\r\n\tat java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.access$201(ScheduledThreadPoolExecutor.java:180)\r\n\tat java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:293)\r\n\tat java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)\r\n\tat java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)\r\n\tat java.lang.Thread.run(Thread.java:745)\r\n```\r\n\r\nWhich indicates that in my lift operator, even when I check for isUnsubscribed it could return false, but immediately afterwards when I call onNext it actually is!\r\n\r\nAny idea how to fix this?"}}