GitHub PullRequestReviewCommentEvent at 2014-10-15 06:52:57
Repository: RxJava
Actor: benjchristensen
Payload: {"action":"created","comment":{"url":"https://api.github.com/repos/ReactiveX/RxJava/pulls/comments/18877173","id":18877173,"diff_hunk":"@@ -15,195 +15,83 @@\n  */\n package rx.internal.operators;\n \n-import java.util.ArrayList;\n-import java.util.List;\n-import java.util.Map;\n-import java.util.WeakHashMap;\n+import java.util.concurrent.atomic.AtomicInteger;\n+import java.util.concurrent.atomic.AtomicReference;\n+\n import rx.Observable.OnSubscribe;\n import rx.Subscriber;\n import rx.Subscription;\n+import rx.exceptions.Exceptions;\n import rx.functions.Action0;\n+import rx.functions.Action1;\n import rx.observables.ConnectableObservable;\n+import rx.subscriptions.MultipleAssignmentSubscription;\n import rx.subscriptions.Subscriptions;\n \n /**\n- * Returns an observable sequence that stays connected to the source as long\n- * as there is at least one subscription to the observable sequence.\n- * @param <T> the value type\n+ * Returns an observable sequence that stays connected to the source as long as\n+ * there is at least one subscription to the observable sequence.\n+ * \n+ * @param <T>\n+ *            the value type\n  */\n public final class OnSubscribeRefCount<T> implements OnSubscribe<T> {\n-    final ConnectableObservable<? extends T> source;\n-    final Object guard;\n-    /** Guarded by guard. */\n-    int index;\n-    /** Guarded by guard. */\n-    boolean emitting;\n-    /** Guarded by guard. If true, indicates a connection request, false indicates a disconnect request. */\n-    List<Token> queue;\n-    /** Manipulated while in the serialized section. */\n-    int count;\n-    /** Manipulated while in the serialized section. */\n-    Subscription connection;\n-    /** Manipulated while in the serialized section. */\n-    final Map<Token, Object> connectionStatus;\n-    /** Occupied indicator. */\n-    private static final Object OCCUPIED = new Object();\n+\n+    private final ConnectableObservable<? extends T> source;\n+    private final AtomicReference<MultipleAssignmentSubscription> connectedSubscription = new AtomicReference<MultipleAssignmentSubscription>();\n+    private final AtomicInteger subscriptionCount = new AtomicInteger(0);\n+\n+    /**\n+     * Constructor.\n+     * \n+     * @param source\n+     *            observable to apply ref count to\n+     */\n     public OnSubscribeRefCount(ConnectableObservable<? extends T> source) {\n         this.source = source;\n-        this.guard = new Object();\n-        this.connectionStatus = new WeakHashMap<Token, Object>();\n     }\n \n     @Override\n-    public void call(Subscriber<? super T> t1) {\n-        int id;\n-        synchronized (guard) {\n-            id = ++index;\n+    public void call(final Subscriber<? super T> subscriber) {\n+        if (subscriber.isUnsubscribed()) {\n+            return;\n+        }\n+        // always subscribe\n+        source.unsafeSubscribe(subscriber);\n+\n+        int count = subscriptionCount.getAndIncrement();\n+        if (count == 0) {\n+            // starting a new connection\n+            final MultipleAssignmentSubscription connectableSubscriptionHolder = new MultipleAssignmentSubscription();","path":"src/main/java/rx/internal/operators/OnSubscribeRefCount.java","position":80,"original_position":80,"commit_id":"696d25d921b4b16c4e98f681c7711c879ffc38a1","original_commit_id":"696d25d921b4b16c4e98f681c7711c879ffc38a1","user":{"login":"benjchristensen","id":813492,"avatar_url":"https://avatars.githubusercontent.com/u/813492?v=2","gravatar_id":"","url":"https://api.github.com/users/benjchristensen","html_url":"https://github.com/benjchristensen","followers_url":"https://api.github.com/users/benjchristensen/followers","following_url":"https://api.github.com/users/benjchristensen/following{/other_user}","gists_url":"https://api.github.com/users/benjchristensen/gists{/gist_id}","starred_url":"https://api.github.com/users/benjchristensen/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/benjchristensen/subscriptions","organizations_url":"https://api.github.com/users/benjchristensen/orgs","repos_url":"https://api.github.com/users/benjchristensen/repos","events_url":"https://api.github.com/users/benjchristensen/events{/privacy}","received_events_url":"https://api.github.com/users/benjchristensen/received_events","type":"User","site_admin":false},"body":"This is not thread-safe because 1 thread could be creating this subscription while a 2nd jumps ahead, registers its subscription then gets a null when it unsubscribes and blows up with an NPE all before the 1st thread finishes creating the MultipleAssignmentSubscription.\r\n\r\nWe either need locks or to use a State object with a single AtomicReference containing them all in a state machine.","created_at":"2014-10-15T06:52:57Z","updated_at":"2014-10-15T06:52:57Z","html_url":"https://github.com/ReactiveX/RxJava/pull/1754#discussion_r18877173","pull_request_url":"https://api.github.com/repos/ReactiveX/RxJava/pulls/1754","_links":{"self":{"href":"https://api.github.com/repos/ReactiveX/RxJava/pulls/comments/18877173"},"html":{"href":"https://github.com/ReactiveX/RxJava/pull/1754#discussion_r18877173"},"pull_request":{"href":"https://api.github.com/repos/ReactiveX/RxJava/pulls/1754"}}},"pull_request":{"url":"https://api.github.com/repos/ReactiveX/RxJava/pulls/1754","id":22763855,"html_url":"https://github.com/ReactiveX/RxJava/pull/1754","diff_url":"https://github.com/ReactiveX/RxJava/pull/1754.diff","patch_url":"https://github.com/ReactiveX/RxJava/pull/1754.patch","issue_url":"https://api.github.com/repos/ReactiveX/RxJava/issues/1754","number":1754,"state":"open","locked":false,"title":"WIP: OnSubscribeRefCount Synchronous","user":{"login":"benjchristensen","id":813492,"avatar_url":"https://avatars.githubusercontent.com/u/813492?v=2","gravatar_id":"","url":"https://api.github.com/users/benjchristensen","html_url":"https://github.com/benjchristensen","followers_url":"https://api.github.com/users/benjchristensen/followers","following_url":"https://api.github.com/users/benjchristensen/following{/other_user}","gists_url":"https://api.github.com/users/benjchristensen/gists{/gist_id}","starred_url":"https://api.github.com/users/benjchristensen/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/benjchristensen/subscriptions","organizations_url":"https://api.github.com/users/benjchristensen/orgs","repos_url":"https://api.github.com/users/benjchristensen/repos","events_url":"https://api.github.com/users/benjchristensen/events{/privacy}","received_events_url":"https://api.github.com/users/benjchristensen/received_events","type":"User","site_admin":false},"body":"This is a work in progress... do not merge or use this.\r\n\r\nSubmitting as PR to allow discussion.","created_at":"2014-10-15T06:50:01Z","updated_at":"2014-10-15T06:52:57Z","closed_at":null,"merged_at":null,"merge_commit_sha":"05f279c215cbc07591a48651938cbec8d4528936","assignee":null,"milestone":null,"commits_url":"https://api.github.com/repos/ReactiveX/RxJava/pulls/1754/commits","review_comments_url":"https://api.github.com/repos/ReactiveX/RxJava/pulls/1754/comments","review_comment_url":"https://api.github.com/repos/ReactiveX/RxJava/pulls/comments/{number}","comments_url":"https://api.github.com/repos/ReactiveX/RxJava/issues/1754/comments","statuses_url":"https://api.github.com/repos/ReactiveX/RxJava/statuses/696d25d921b4b16c4e98f681c7711c879ffc38a1","head":{"label":"benjchristensen:issue1688-refCount","ref":"issue1688-refCount","sha":"696d25d921b4b16c4e98f681c7711c879ffc38a1","user":{"login":"benjchristensen","id":813492,"avatar_url":"https://avatars.githubusercontent.com/u/813492?v=2","gravatar_id":"","url":"https://api.github.com/users/benjchristensen","html_url":"https://github.com/benjchristensen","followers_url":"https://api.github.com/users/benjchristensen/followers","following_url":"https://api.github.com/users/benjchristensen/following{/other_user}","gists_url":"https://api.github.com/users/benjchristensen/gists{/gist_id}","starred_url":"https://api.github.com/users/benjchristensen/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/benjchristensen/subscriptions","organizations_url":"https://api.github.com/users/benjchristensen/orgs","repos_url":"https://api.github.com/users/benjchristensen/repos","events_url":"https://api.github.com/users/benjchristensen/events{/privacy}","received_events_url":"https://api.github.com/users/benjchristensen/received_events","type":"User","site_admin":false},"repo":{"id":7545816,"name":"RxJava","full_name":"benjchristensen/RxJava","owner":{"login":"benjchristensen","id":813492,"avatar_url":"https://avatars.githubusercontent.com/u/813492?v=2","gravatar_id":"","url":"https://api.github.com/users/benjchristensen","html_url":"https://github.com/benjchristensen","followers_url":"https://api.github.com/users/benjchristensen/followers","following_url":"https://api.github.com/users/benjchristensen/following{/other_user}","gists_url":"https://api.github.com/users/benjchristensen/gists{/gist_id}","starred_url":"https://api.github.com/users/benjchristensen/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/benjchristensen/subscriptions","organizations_url":"https://api.github.com/users/benjchristensen/orgs","repos_url":"https://api.github.com/users/benjchristensen/repos","events_url":"https://api.github.com/users/benjchristensen/events{/privacy}","received_events_url":"https://api.github.com/users/benjchristensen/received_events","type":"User","site_admin":false},"private":false,"html_url":"https://github.com/benjchristensen/RxJava","description":"","fork":true,"url":"https://api.github.com/repos/benjchristensen/RxJava","forks_url":"https://api.github.com/repos/benjchristensen/RxJava/forks","keys_url":"https://api.github.com/repos/benjchristensen/RxJava/keys{/key_id}","collaborators_url":"https://api.github.com/repos/benjchristensen/RxJava/collaborators{/collaborator}","teams_url":"https://api.github.com/repos/benjchristensen/RxJava/teams","hooks_url":"https://api.github.com/repos/benjchristensen/RxJava/hooks","issue_events_url":"https://api.github.com/repos/benjchristensen/RxJava/issues/events{/number}","events_url":"https://api.github.com/repos/benjchristensen/RxJava/events","assignees_url":"https://api.github.com/repos/benjchristensen/RxJava/assignees{/user}","branches_url":"https://api.github.com/repos/benjchristensen/RxJava/branches{/branch}","tags_url":"https://api.github.com/repos/benjchristensen/RxJava/tags","blobs_url":"https://api.github.com/repos/benjchristensen/RxJava/git/blobs{/sha}","git_tags_url":"https://api.github.com/repos/benjchristensen/RxJava/git/tags{/sha}","git_refs_url":"https://api.github.com/repos/benjchristensen/RxJava/git/refs{/sha}","trees_url":"https://api.github.com/repos/benjchristensen/RxJava/git/trees{/sha}","statuses_url":"https://api.github.com/repos/benjchristensen/RxJava/statuses/{sha}","languages_url":"https://api.github.com/repos/benjchristensen/RxJava/languages","stargazers_url":"https://api.github.com/repos/benjchristensen/RxJava/stargazers","contributors_url":"https://api.github.com/repos/benjchristensen/RxJava/contributors","subscribers_url":"https://api.github.com/repos/benjchristensen/RxJava/subscribers","subscription_url":"https://api.github.com/repos/benjchristensen/RxJava/subscription","commits_url":"https://api.github.com/repos/benjchristensen/RxJava/commits{/sha}","git_commits_url":"https://api.github.com/repos/benjchristensen/RxJava/git/commits{/sha}","comments_url":"https://api.github.com/repos/benjchristensen/RxJava/comments{/number}","issue_comment_url":"https://api.github.com/repos/benjchristensen/RxJava/issues/comments/{number}","contents_url":"https://api.github.com/repos/benjchristensen/RxJava/contents/{+path}","compare_url":"https://api.github.com/repos/benjchristensen/RxJava/compare/{base}...{head}","merges_url":"https://api.github.com/repos/benjchristensen/RxJava/merges","archive_url":"https://api.github.com/repos/benjchristensen/RxJava/{archive_format}{/ref}","downloads_url":"https://api.github.com/repos/benjchristensen/RxJava/downloads","issues_url":"https://api.github.com/repos/benjchristensen/RxJava/issues{/number}","pulls_url":"https://api.github.com/repos/benjchristensen/RxJava/pulls{/number}","milestones_url":"https://api.github.com/repos/benjchristensen/RxJava/milestones{/number}","notifications_url":"https://api.github.com/repos/benjchristensen/RxJava/notifications{?since,all,participating}","labels_url":"https://api.github.com/repos/benjchristensen/RxJava/labels{/name}","releases_url":"https://api.github.com/repos/benjchristensen/RxJava/releases{/id}","created_at":"2013-01-10T17:58:51Z","updated_at":"2014-07-02T16:36:06Z","pushed_at":"2014-10-15T06:44:13Z","git_url":"git://github.com/benjchristensen/RxJava.git","ssh_url":"git@github.com:benjchristensen/RxJava.git","clone_url":"https://github.com/benjchristensen/RxJava.git","svn_url":"https://github.com/benjchristensen/RxJava","homepage":null,"size":21869,"stargazers_count":0,"watchers_count":0,"language":"Java","has_issues":false,"has_downloads":true,"has_wiki":true,"has_pages":false,"forks_count":0,"mirror_url":null,"open_issues_count":0,"forks":0,"open_issues":0,"watchers":0,"default_branch":"master"}},"base":{"label":"ReactiveX:1.x","ref":"1.x","sha":"dd389126965be72c6560b7dd0bd5fbbba1443da4","user":{"login":"ReactiveX","id":6407041,"avatar_url":"https://avatars.githubusercontent.com/u/6407041?v=2","gravatar_id":"","url":"https://api.github.com/users/ReactiveX","html_url":"https://github.com/ReactiveX","followers_url":"https://api.github.com/users/ReactiveX/followers","following_url":"https://api.github.com/users/ReactiveX/following{/other_user}","gists_url":"https://api.github.com/users/ReactiveX/gists{/gist_id}","starred_url":"https://api.github.com/users/ReactiveX/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/ReactiveX/subscriptions","organizations_url":"https://api.github.com/users/ReactiveX/orgs","repos_url":"https://api.github.com/users/ReactiveX/repos","events_url":"https://api.github.com/users/ReactiveX/events{/privacy}","received_events_url":"https://api.github.com/users/ReactiveX/received_events","type":"Organization","site_admin":false},"repo":{"id":7508411,"name":"RxJava","full_name":"ReactiveX/RxJava","owner":{"login":"ReactiveX","id":6407041,"avatar_url":"https://avatars.githubusercontent.com/u/6407041?v=2","gravatar_id":"","url":"https://api.github.com/users/ReactiveX","html_url":"https://github.com/ReactiveX","followers_url":"https://api.github.com/users/ReactiveX/followers","following_url":"https://api.github.com/users/ReactiveX/following{/other_user}","gists_url":"https://api.github.com/users/ReactiveX/gists{/gist_id}","starred_url":"https://api.github.com/users/ReactiveX/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/ReactiveX/subscriptions","organizations_url":"https://api.github.com/users/ReactiveX/orgs","repos_url":"https://api.github.com/users/ReactiveX/repos","events_url":"https://api.github.com/users/ReactiveX/events{/privacy}","received_events_url":"https://api.github.com/users/ReactiveX/received_events","type":"Organization","site_admin":false},"private":false,"html_url":"https://github.com/ReactiveX/RxJava","description":"RxJava – Reactive Extensions for the JVM – a library for composing asynchronous and event-based programs using observable sequences for the Java VM.","fork":false,"url":"https://api.github.com/repos/ReactiveX/RxJava","forks_url":"https://api.github.com/repos/ReactiveX/RxJava/forks","keys_url":"https://api.github.com/repos/ReactiveX/RxJava/keys{/key_id}","collaborators_url":"https://api.github.com/repos/ReactiveX/RxJava/collaborators{/collaborator}","teams_url":"https://api.github.com/repos/ReactiveX/RxJava/teams","hooks_url":"https://api.github.com/repos/ReactiveX/RxJava/hooks","issue_events_url":"https://api.github.com/repos/ReactiveX/RxJava/issues/events{/number}","events_url":"https://api.github.com/repos/ReactiveX/RxJava/events","assignees_url":"https://api.github.com/repos/ReactiveX/RxJava/assignees{/user}","branches_url":"https://api.github.com/repos/ReactiveX/RxJava/branches{/branch}","tags_url":"https://api.github.com/repos/ReactiveX/RxJava/tags","blobs_url":"https://api.github.com/repos/ReactiveX/RxJava/git/blobs{/sha}","git_tags_url":"https://api.github.com/repos/ReactiveX/RxJava/git/tags{/sha}","git_refs_url":"https://api.github.com/repos/ReactiveX/RxJava/git/refs{/sha}","trees_url":"https://api.github.com/repos/ReactiveX/RxJava/git/trees{/sha}","statuses_url":"https://api.github.com/repos/ReactiveX/RxJava/statuses/{sha}","languages_url":"https://api.github.com/repos/ReactiveX/RxJava/languages","stargazers_url":"https://api.github.com/repos/ReactiveX/RxJava/stargazers","contributors_url":"https://api.github.com/repos/ReactiveX/RxJava/contributors","subscribers_url":"https://api.github.com/repos/ReactiveX/RxJava/subscribers","subscription_url":"https://api.github.com/repos/ReactiveX/RxJava/subscription","commits_url":"https://api.github.com/repos/ReactiveX/RxJava/commits{/sha}","git_commits_url":"https://api.github.com/repos/ReactiveX/RxJava/git/commits{/sha}","comments_url":"https://api.github.com/repos/ReactiveX/RxJava/comments{/number}","issue_comment_url":"https://api.github.com/repos/ReactiveX/RxJava/issues/comments/{number}","contents_url":"https://api.github.com/repos/ReactiveX/RxJava/contents/{+path}","compare_url":"https://api.github.com/repos/ReactiveX/RxJava/compare/{base}...{head}","merges_url":"https://api.github.com/repos/ReactiveX/RxJava/merges","archive_url":"https://api.github.com/repos/ReactiveX/RxJava/{archive_format}{/ref}","downloads_url":"https://api.github.com/repos/ReactiveX/RxJava/downloads","issues_url":"https://api.github.com/repos/ReactiveX/RxJava/issues{/number}","pulls_url":"https://api.github.com/repos/ReactiveX/RxJava/pulls{/number}","milestones_url":"https://api.github.com/repos/ReactiveX/RxJava/milestones{/number}","notifications_url":"https://api.github.com/repos/ReactiveX/RxJava/notifications{?since,all,participating}","labels_url":"https://api.github.com/repos/ReactiveX/RxJava/labels{/name}","releases_url":"https://api.github.com/repos/ReactiveX/RxJava/releases{/id}","created_at":"2013-01-08T20:11:48Z","updated_at":"2014-10-15T01:50:40Z","pushed_at":"2014-10-14T20:39:56Z","git_url":"git://github.com/ReactiveX/RxJava.git","ssh_url":"git@github.com:ReactiveX/RxJava.git","clone_url":"https://github.com/ReactiveX/RxJava.git","svn_url":"https://github.com/ReactiveX/RxJava","homepage":"","size":637439,"stargazers_count":3374,"watchers_count":3374,"language":"Java","has_issues":true,"has_downloads":true,"has_wiki":true,"has_pages":true,"forks_count":511,"mirror_url":null,"open_issues_count":52,"forks":511,"open_issues":52,"watchers":3374,"default_branch":"1.x"}},"_links":{"self":{"href":"https://api.github.com/repos/ReactiveX/RxJava/pulls/1754"},"html":{"href":"https://github.com/ReactiveX/RxJava/pull/1754"},"issue":{"href":"https://api.github.com/repos/ReactiveX/RxJava/issues/1754"},"comments":{"href":"https://api.github.com/repos/ReactiveX/RxJava/issues/1754/comments"},"review_comments":{"href":"https://api.github.com/repos/ReactiveX/RxJava/pulls/1754/comments"},"review_comment":{"href":"https://api.github.com/repos/ReactiveX/RxJava/pulls/comments/{number}"},"commits":{"href":"https://api.github.com/repos/ReactiveX/RxJava/pulls/1754/commits"},"statuses":{"href":"https://api.github.com/repos/ReactiveX/RxJava/statuses/696d25d921b4b16c4e98f681c7711c879ffc38a1"}}}}