{"action":"created","comment":{"url":"https://api.github.com/repos/ReactiveX/RxJava/pulls/comments/32684801","id":32684801,"diff_hunk":"@@ -71,715 +72,878 @@\n         }\n         return (OperatorMerge<T>)HolderNoDelay.INSTANCE;\n     }\n-    /*\n-     * benjchristensen => This class is complex and I'm not a fan of it despite writing it. I want to give some background\n-     * as to why for anyone who wants to try and help improve it.\n-     * \n-     * One of my first implementations that added backpressure support (Producer.request) was fairly elegant and used a simple\n-     * queue draining approach. It was simple to understand as all onNext were added to their queues, then a single winner\n-     * would drain the queues, similar to observeOn. It killed the Netflix API when I canaried it. There were two problems:\n-     * (1) performance and (2) object allocation overhead causing massive GC pressure. Remember that merge is one of the most\n-     * used operators (mostly due to flatmap) and is therefore critical to and a limiter of performance in any application.\n-     * \n-     * All subsequent work on this class and the various fast-paths and branches within it have been to achieve the needed functionality\n-     * while reducing or eliminating object allocation and keeping performance acceptable.\n-     * \n-     * This has meant adopting strategies such as:\n-     * \n-     * - ring buffers instead of growable queues\n-     * - object pooling\n-     * - skipping request logic when downstream does not need backpressure\n-     * - ScalarValueQueue for optimizing synchronous single-value Observables\n-     * - adopting data structures that use Unsafe (and gating them based on environment so non-Oracle JVMs still work)\n-     * \n-     * It has definitely increased the complexity and maintenance cost of this class, but the performance gains have been significant.\n-     * \n-     * The biggest cost of the increased complexity is concurrency bugs and reasoning through what's going on.\n-     * \n-     * I'd love to have contributions that improve this class, but keep in mind the performance and GC pressure.\n-     * The benchmarks I use are in the JMH OperatorMergePerf class. GC memory pressure is tested using Java Flight Recorder\n-     * to track object allocation.\n+    /**\n+     * Creates a new instance of the operator with the given delayError and maxConcurrency settings.\n+     * @param delayErrors\n+     * @param maxConcurrent the maximum number of concurrent subscriptions or Integer.MAX_VALUE for unlimited\n+     * @return\n      */\n-\n-    private OperatorMerge() {\n-        this.delayErrors = false;\n+    public static <T> OperatorMerge<T> instance(boolean delayErrors, int maxConcurrent) {\n+        if (maxConcurrent == Integer.MAX_VALUE) {\n+            return instance(delayErrors);\n+        }\n+        return new OperatorMerge<T>(delayErrors, maxConcurrent);\n     }\n \n-    private OperatorMerge(boolean delayErrors) {\n+    final boolean delayErrors;\n+    final int maxConcurrent;\n+\n+    private OperatorMerge(boolean delayErrors, int maxConcurrent) {\n         this.delayErrors = delayErrors;\n+        this.maxConcurrent = maxConcurrent;\n     }\n \n-    private final boolean delayErrors;\n-\n     @Override\n     public Subscriber<Observable<? extends T>> call(final Subscriber<? super T> child) {\n-        return new MergeSubscriber<T>(child, delayErrors);\n-\n+        MergeSubscriber<T> subscriber = new MergeSubscriber<T>(child, delayErrors, maxConcurrent);\n+        MergeProducer<T> producer = new MergeProducer<T>(subscriber);\n+        subscriber.producer = producer;\n+        producer.lazySet(0); // make sure subscriber.producer is visible\n+        \n+        child.add(subscriber);\n+        child.setProducer(producer);\n+        \n+        return subscriber;\n     }\n \n-    private static final class MergeSubscriber<T> extends Subscriber<Observable<? extends T>> {\n-        final NotificationLite<T> on = NotificationLite.instance();\n-        final Subscriber<? super T> actual;\n-        private final MergeProducer<T> mergeProducer;\n-        private int wip;\n-        private boolean completed;\n-        private final boolean delayErrors;\n-        private ConcurrentLinkedQueue<Throwable> exceptions;\n-\n-        private volatile SubscriptionIndexedRingBuffer<InnerSubscriber<T>> childrenSubscribers;\n-\n-        private volatile RxRingBuffer scalarValueQueue = null;\n-\n-        /* protected by lock on MergeSubscriber instance */\n-        private int missedEmitting = 0;\n-        private boolean emitLock = false;\n+    static final class MergeProducer<T> extends AtomicLong implements Producer {\n+        /** */\n+        private static final long serialVersionUID = -1214379189873595503L;\n \n-        /**\n-         * Using synchronized(this) for `emitLock` instead of ReentrantLock or AtomicInteger is faster when there is no contention.\n-         * \n-         * <pre> {@code\n-         * Using ReentrantLock:\n-         * r.o.OperatorMergePerf.merge1SyncStreamOfN           1000  thrpt         5    44185.294     1295.565    ops/s\n-         * \n-         * Using synchronized(this):\n-         * r.o.OperatorMergePerf.merge1SyncStreamOfN           1000  thrpt         5    79715.981     3704.486    ops/s\n-         * \n-         * Still slower though than allowing concurrency:\n-         * r.o.OperatorMergePerf.merge1SyncStreamOfN           1000  thrpt         5   149331.046     4851.290    ops/s\n-         * } </pre>\n-         */\n-\n-        public MergeSubscriber(Subscriber<? super T> actual, boolean delayErrors) {\n-            super(actual);\n-            this.actual = actual;\n-            this.mergeProducer = new MergeProducer<T>(this);\n-            this.delayErrors = delayErrors;\n-            // decoupled the subscription chain because we need to decouple and control backpressure\n-            actual.add(this);\n-            actual.setProducer(mergeProducer);\n-        }\n-\n-        @Override\n-        public void onStart() {\n-            // we request backpressure so we can handle long-running Observables that are enqueueing, such as flatMap use cases\n-            // we decouple the Producer chain while keeping the Subscription chain together (perf benefit) via super(actual)\n-            request(RxRingBuffer.SIZE);\n+        final MergeSubscriber<T> subscriber;\n+        /** Indicates the producer mode: -1 no request, 0 unbounded fast, 1 bounded. */\n+        volatile int mode;\n+        @SuppressWarnings(\"rawtypes\")\n+        static final AtomicIntegerFieldUpdater<MergeProducer> MODE\n+        = AtomicIntegerFieldUpdater.newUpdater(MergeProducer.class, \"mode\");\n+        \n+        public MergeProducer(MergeSubscriber<T> subscriber) {\n+            this.subscriber = subscriber;\n+            this.mode = -1;\n         }\n-\n-        /*\n-         * This is expected to be executed sequentially as per the Rx contract or it will not work.\n-         */\n+        \n         @Override\n-        public void onNext(Observable<? extends T> t) {\n-            if (t instanceof ScalarSynchronousObservable) {\n-                ScalarSynchronousObservable<? extends T> t2 = (ScalarSynchronousObservable<? extends T>)t;\n-                handleScalarSynchronousObservable(t2);\n-            } else {\n-                if (t == null || isUnsubscribed()) {\n+        public void request(long n) {\n+            if (n >= 0) {\n+                int m = mode;\n+                // if in unbounded mode, quit\n+                if (m == 0) {\n                     return;\n                 }\n-                synchronized (this) {\n-                    // synchronized here because `wip` can be concurrently changed by children Observables\n-                    wip++;\n+                // if bounded mode, update request count and emit\n+                if (m == 1) {\n+                    BackpressureUtils.getAndAddRequest(this, n);\n+                    subscriber.emit();\n+                    return;\n+                }\n+                // mode has not been set\n+                if (n == Long.MAX_VALUE) {\n+                    // try setting mode to unbounded\n+                    if (MODE.compareAndSet(this, -1, 0)) {\n+                        subscriber.requestAll();\n+                        return;\n+                    }\n+                } else {\n+                    if (MODE.compareAndSet(this, -1, 1)) {\n+                        BackpressureUtils.getAndAddRequest(this, n);\n+                        subscriber.requestDefault();\n+                        return;\n+                    }\n                 }\n-                handleNewSource(t);\n+                // mode has changed concurrently\n+                // recheck state\n+                m = mode;\n+                // if in unbounded mode, quit\n+                if (m == 0) {\n+                    return;\n+                } else\n+                // if bounded mode, update request count and emit\n+                if (m == 1) {\n+                    BackpressureUtils.getAndAddRequest(this, n);\n+                    subscriber.emit();\n+                    return;\n+                }\n+                throw new IllegalStateException(\"Shouldn't be in mode: \" + m);\n             }\n         }\n-\n-        private void handleNewSource(Observable<? extends T> t) {\n-            if (childrenSubscribers == null) {\n-                // lazily create this only if we receive Observables we need to subscribe to\n-                childrenSubscribers = new SubscriptionIndexedRingBuffer<InnerSubscriber<T>>();\n-                add(childrenSubscribers);\n-            }\n-            MergeProducer<T> producerIfNeeded = null;\n-            // if we have received a request then we need to respect it, otherwise we fast-path\n-            if (mergeProducer.requested != Long.MAX_VALUE) {\n-                /**\n-                 * <pre> {@code\n-                 * With this optimization:\n-                 * \n-                 * r.o.OperatorMergePerf.merge1SyncStreamOfN      1000  thrpt         5    57100.080     4686.331    ops/s\n-                 * r.o.OperatorMergePerf.merge1SyncStreamOfN   1000000  thrpt         5       60.875        1.622    ops/s\n-                 *  \n-                 * Without this optimization:\n-                 * \n-                 * r.o.OperatorMergePerf.merge1SyncStreamOfN      1000  thrpt         5    29863.945     1858.002    ops/s\n-                 * r.o.OperatorMergePerf.merge1SyncStreamOfN   1000000  thrpt         5       30.516        1.087    ops/s\n-                 * } </pre>\n-                 */\n-                producerIfNeeded = mergeProducer;\n-            }\n-            InnerSubscriber<T> i = new InnerSubscriber<T>(this, producerIfNeeded);\n-            i.sindex = childrenSubscribers.add(i);\n-            t.unsafeSubscribe(i);\n-            if (!isUnsubscribed()) {\n-                request(1);\n-            }\n+        public long produced(int n) {\n+            return addAndGet(-n);\n         }\n+    }\n+    \n+    /**\n+     * The subscriber that observes Observables. \n+     * @param <T> the value type\n+     */\n+    static final class MergeSubscriber<T> extends Subscriber<Observable<? extends T>> {\n+        final Subscriber<? super T> child;\n+        final boolean delayErrors;\n+        final int maxConcurrent;\n+        \n+        MergeProducer<T> producer;\n+        \n+        /** Upstream might not produce scalar values at all so avoid creating a queue for it upfront. */\n+        volatile RxRingBuffer scalarQueue;\n+        /** Tracks the active subscriptions to sources. */\n+        volatile CompositeSubscription subscriptions;\n+        /** Due to the emission loop, we need to store errors somewhere if !delayErrors. */\n+        volatile ConcurrentLinkedQueue<Throwable> errors;\n+        \n+        final NotificationLite<T> nl;\n+        \n+        volatile boolean done;\n+        \n+        /** Guarded by this. */\n+        boolean emitting;\n+        /** Guarded by this. */\n+        boolean missed;\n+        \n+        final Object innerGuard;\n+        /** Copy-on-write array, guarded by innerGuard. */\n+        volatile InnerSubscriber<?>[] innerSubscribers;\n+        \n+        /** Used to generate unique InnerSubscriber IDs. Modified from onNext only. */\n+        long uniqueId;\n+        \n+        /** Which was the last InnerSubscriber that emitted? Accessed if emitting == true. */\n+        long lastId;\n+        /** What was its index in the innerSubscribers array? Accessed if emitting == true. */\n+        int lastIndex;\n+        \n+        /** An empty array to avoid creating new empty arrays in removeInner. */ \n+        static final InnerSubscriber<?>[] EMPTY = new InnerSubscriber<?>[0];\n \n-        private void handleScalarSynchronousObservable(ScalarSynchronousObservable<? extends T> t) {\n-            // fast-path for scalar, synchronous values such as Observable.from(int)\n-            /**\n-             * Without this optimization:\n-             * \n-             * <pre> {@code\n-             * Benchmark                                          (size)   Mode   Samples        Score  Score error    Units\n-             * r.o.OperatorMergePerf.oneStreamOfNthatMergesIn1         1  thrpt         5  2,418,452.409   130572.665    ops/s\n-             * r.o.OperatorMergePerf.oneStreamOfNthatMergesIn1      1000  thrpt         5     5,690.456       94.958    ops/s\n-             * r.o.OperatorMergePerf.merge1SyncStreamOfN         1000000  thrpt         5          takes too long\n-             * \n-             * With this optimization:\n-             * \n-             * r.o.OperatorMergePerf.merge1SyncStreamOfN               1  thrpt         5  5,475,300.198   156741.334    ops/s\n-             * r.o.OperatorMergePerf.merge1SyncStreamOfN            1000  thrpt         5    68,932.278     1311.023    ops/s\n-             * r.o.OperatorMergePerf.merge1SyncStreamOfN         1000000  thrpt         5       64.405        0.611    ops/s\n-             * } </pre>\n-             * \n-             */\n-            if (mergeProducer.requested == Long.MAX_VALUE) {\n-                handleScalarSynchronousObservableWithoutRequestLimits(t);\n-            } else {\n-                handleScalarSynchronousObservableWithRequestLimits(t);\n+        /** In fast mode, this counts the number of active subscriptions, including the one to upstream. */\n+        volatile int fastWip;\n+        @SuppressWarnings(\"rawtypes\")\n+        static final AtomicIntegerFieldUpdater<MergeSubscriber> FAST_WIP\n+        = AtomicIntegerFieldUpdater.newUpdater(MergeSubscriber.class, \"fastWip\");\n+        \n+        volatile boolean fastMode;\n+        \n+        /** Guarded by this. */\n+        Object[] fastArray;\n+        /** Guarded by this. */\n+        int fastSize;\n+        \n+        /** Fast, unbounded emission report contents of 'errors'. */\n+        static final Object REPORT_ERROR = new Object();\n+        /** Fast, unbounded emission complete. */\n+        static final Object REPORT_COMPLETED = new Object();\n+        \n+        public MergeSubscriber(Subscriber<? super T> child, boolean delayErrors, int maxConcurrent) {\n+            this.child = child;\n+            this.delayErrors = delayErrors;\n+            this.maxConcurrent = maxConcurrent;\n+            this.nl = NotificationLite.instance();\n+            this.innerGuard = new Object();\n+            this.innerSubscribers = EMPTY;\n+            this.request(0);","path":"src/main/java/rx/internal/operators/OperatorMerge.java","position":359,"original_position":359,"commit_id":"8a6d506bb199d439d72a9b884cdda624d178a4db","original_commit_id":"8a6d506bb199d439d72a9b884cdda624d178a4db","user":{"login":"benjchristensen","id":813492,"avatar_url":"https://avatars.githubusercontent.com/u/813492?v=3","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":"Curious as to why you are calling this here instead of using `onStart`?\r\n\r\nI think this is safe to do from within the constructor due to when this all gets initialized, but on first read it felt wrong for this to happen here. Probably not an issue but would like to understand your thinking for not following the normal pattern of using `onStart`. ","created_at":"2015-06-17T23:01:32Z","updated_at":"2015-06-17T23:01:32Z","html_url":"https://github.com/ReactiveX/RxJava/pull/2928#discussion_r32684801","pull_request_url":"https://api.github.com/repos/ReactiveX/RxJava/pulls/2928","_links":{"self":{"href":"https://api.github.com/repos/ReactiveX/RxJava/pulls/comments/32684801"},"html":{"href":"https://github.com/ReactiveX/RxJava/pull/2928#discussion_r32684801"},"pull_request":{"href":"https://api.github.com/repos/ReactiveX/RxJava/pulls/2928"}}},"pull_request":{"url":"https://api.github.com/repos/ReactiveX/RxJava/pulls/2928","id":34495543,"html_url":"https://github.com/ReactiveX/RxJava/pull/2928","diff_url":"https://github.com/ReactiveX/RxJava/pull/2928.diff","patch_url":"https://github.com/ReactiveX/RxJava/pull/2928.patch","issue_url":"https://api.github.com/repos/ReactiveX/RxJava/issues/2928","number":2928,"state":"open","locked":false,"title":"Merge and MergeMaxConcurrent unified and rewritten","user":{"login":"akarnokd","id":1269832,"avatar_url":"https://avatars.githubusercontent.com/u/1269832?v=3","gravatar_id":"","url":"https://api.github.com/users/akarnokd","html_url":"https://github.com/akarnokd","followers_url":"https://api.github.com/users/akarnokd/followers","following_url":"https://api.github.com/users/akarnokd/following{/other_user}","gists_url":"https://api.github.com/users/akarnokd/gists{/gist_id}","starred_url":"https://api.github.com/users/akarnokd/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/akarnokd/subscriptions","organizations_url":"https://api.github.com/users/akarnokd/orgs","repos_url":"https://api.github.com/users/akarnokd/repos","events_url":"https://api.github.com/users/akarnokd/events{/privacy}","received_events_url":"https://api.github.com/users/akarnokd/received_events","type":"User","site_admin":false},"body":"I've rewritten ```merge``` and ```mergeMaxConcurrent``` together from scratch.\r\n\r\n  - Added experimental ```maxConcurrent``` overloads to the array-merge and the mergeDelayError\r\n  - Fixed ```OperatorMapNotification``` beause it disregarded backpressure on its terminal emissions.\r\n  - While debugging some merge test failures, I've encountered a bug in OperatorPublish with the same underlying logical error (see comment) and fixed it here.\r\n  - Accidentally introduced a few tabs, therefore, I went in and replaced all tabs with spaces.\r\n  - I've changed ```OperatorMergeTest.shouldNotCompleteIfThereArePendingScalarSynchronousEmissionsWhenTheLastInnerSubscriberCompletes``` because it is legal to get onCompleted without any request.\r\n","created_at":"2015-04-30T20:02:47Z","updated_at":"2015-06-17T23:01:32Z","closed_at":null,"merged_at":null,"merge_commit_sha":"78cb17b3bd8cf7f93def255afaaebece2fbfabaf","assignee":null,"milestone":null,"commits_url":"https://api.github.com/repos/ReactiveX/RxJava/pulls/2928/commits","review_comments_url":"https://api.github.com/repos/ReactiveX/RxJava/pulls/2928/comments","review_comment_url":"https://api.github.com/repos/ReactiveX/RxJava/pulls/comments{/number}","comments_url":"https://api.github.com/repos/ReactiveX/RxJava/issues/2928/comments","statuses_url":"https://api.github.com/repos/ReactiveX/RxJava/statuses/8a6d506bb199d439d72a9b884cdda624d178a4db","head":{"label":"akarnokd:OperatorMergeFullRewrite","ref":"OperatorMergeFullRewrite","sha":"8a6d506bb199d439d72a9b884cdda624d178a4db","user":{"login":"akarnokd","id":1269832,"avatar_url":"https://avatars.githubusercontent.com/u/1269832?v=3","gravatar_id":"","url":"https://api.github.com/users/akarnokd","html_url":"https://github.com/akarnokd","followers_url":"https://api.github.com/users/akarnokd/followers","following_url":"https://api.github.com/users/akarnokd/following{/other_user}","gists_url":"https://api.github.com/users/akarnokd/gists{/gist_id}","starred_url":"https://api.github.com/users/akarnokd/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/akarnokd/subscriptions","organizations_url":"https://api.github.com/users/akarnokd/orgs","repos_url":"https://api.github.com/users/akarnokd/repos","events_url":"https://api.github.com/users/akarnokd/events{/privacy}","received_events_url":"https://api.github.com/users/akarnokd/received_events","type":"User","site_admin":false},"repo":{"id":14518033,"name":"RxJava","full_name":"akarnokd/RxJava","owner":{"login":"akarnokd","id":1269832,"avatar_url":"https://avatars.githubusercontent.com/u/1269832?v=3","gravatar_id":"","url":"https://api.github.com/users/akarnokd","html_url":"https://github.com/akarnokd","followers_url":"https://api.github.com/users/akarnokd/followers","following_url":"https://api.github.com/users/akarnokd/following{/other_user}","gists_url":"https://api.github.com/users/akarnokd/gists{/gist_id}","starred_url":"https://api.github.com/users/akarnokd/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/akarnokd/subscriptions","organizations_url":"https://api.github.com/users/akarnokd/orgs","repos_url":"https://api.github.com/users/akarnokd/repos","events_url":"https://api.github.com/users/akarnokd/events{/privacy}","received_events_url":"https://api.github.com/users/akarnokd/received_events","type":"User","site_admin":false},"private":false,"html_url":"https://github.com/akarnokd/RxJava","description":"RxJava - a library for composing asynchronous and event-based programs using observable sequences for the Java VM.","fork":true,"url":"https://api.github.com/repos/akarnokd/RxJava","forks_url":"https://api.github.com/repos/akarnokd/RxJava/forks","keys_url":"https://api.github.com/repos/akarnokd/RxJava/keys{/key_id}","collaborators_url":"https://api.github.com/repos/akarnokd/RxJava/collaborators{/collaborator}","teams_url":"https://api.github.com/repos/akarnokd/RxJava/teams","hooks_url":"https://api.github.com/repos/akarnokd/RxJava/hooks","issue_events_url":"https://api.github.com/repos/akarnokd/RxJava/issues/events{/number}","events_url":"https://api.github.com/repos/akarnokd/RxJava/events","assignees_url":"https://api.github.com/repos/akarnokd/RxJava/assignees{/user}","branches_url":"https://api.github.com/repos/akarnokd/RxJava/branches{/branch}","tags_url":"https://api.github.com/repos/akarnokd/RxJava/tags","blobs_url":"https://api.github.com/repos/akarnokd/RxJava/git/blobs{/sha}","git_tags_url":"https://api.github.com/repos/akarnokd/RxJava/git/tags{/sha}","git_refs_url":"https://api.github.com/repos/akarnokd/RxJava/git/refs{/sha}","trees_url":"https://api.github.com/repos/akarnokd/RxJava/git/trees{/sha}","statuses_url":"https://api.github.com/repos/akarnokd/RxJava/statuses/{sha}","languages_url":"https://api.github.com/repos/akarnokd/RxJava/languages","stargazers_url":"https://api.github.com/repos/akarnokd/RxJava/stargazers","contributors_url":"https://api.github.com/repos/akarnokd/RxJava/contributors","subscribers_url":"https://api.github.com/repos/akarnokd/RxJava/subscribers","subscription_url":"https://api.github.com/repos/akarnokd/RxJava/subscription","commits_url":"https://api.github.com/repos/akarnokd/RxJava/commits{/sha}","git_commits_url":"https://api.github.com/repos/akarnokd/RxJava/git/commits{/sha}","comments_url":"https://api.github.com/repos/akarnokd/RxJava/comments{/number}","issue_comment_url":"https://api.github.com/repos/akarnokd/RxJava/issues/comments{/number}","contents_url":"https://api.github.com/repos/akarnokd/RxJava/contents/{+path}","compare_url":"https://api.github.com/repos/akarnokd/RxJava/compare/{base}...{head}","merges_url":"https://api.github.com/repos/akarnokd/RxJava/merges","archive_url":"https://api.github.com/repos/akarnokd/RxJava/{archive_format}{/ref}","downloads_url":"https://api.github.com/repos/akarnokd/RxJava/downloads","issues_url":"https://api.github.com/repos/akarnokd/RxJava/issues{/number}","pulls_url":"https://api.github.com/repos/akarnokd/RxJava/pulls{/number}","milestones_url":"https://api.github.com/repos/akarnokd/RxJava/milestones{/number}","notifications_url":"https://api.github.com/repos/akarnokd/RxJava/notifications{?since,all,participating}","labels_url":"https://api.github.com/repos/akarnokd/RxJava/labels{/name}","releases_url":"https://api.github.com/repos/akarnokd/RxJava/releases{/id}","created_at":"2013-11-19T08:04:52Z","updated_at":"2014-07-09T07:02:30Z","pushed_at":"2015-06-17T12:52:02Z","git_url":"git://github.com/akarnokd/RxJava.git","ssh_url":"git@github.com:akarnokd/RxJava.git","clone_url":"https://github.com/akarnokd/RxJava.git","svn_url":"https://github.com/akarnokd/RxJava","homepage":"","size":174977,"stargazers_count":1,"watchers_count":1,"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":1,"default_branch":"master"}},"base":{"label":"ReactiveX:1.x","ref":"1.x","sha":"32179c631d18e120b4d8e70984cb032a4b551c63","user":{"login":"ReactiveX","id":6407041,"avatar_url":"https://avatars.githubusercontent.com/u/6407041?v=3","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=3","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":"2015-06-17T19:45:13Z","pushed_at":"2015-06-17T17:33:30Z","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":520514,"stargazers_count":6103,"watchers_count":6103,"language":"Java","has_issues":true,"has_downloads":true,"has_wiki":true,"has_pages":true,"forks_count":839,"mirror_url":null,"open_issues_count":131,"forks":839,"open_issues":131,"watchers":6103,"default_branch":"1.x"}},"_links":{"self":{"href":"https://api.github.com/repos/ReactiveX/RxJava/pulls/2928"},"html":{"href":"https://github.com/ReactiveX/RxJava/pull/2928"},"issue":{"href":"https://api.github.com/repos/ReactiveX/RxJava/issues/2928"},"comments":{"href":"https://api.github.com/repos/ReactiveX/RxJava/issues/2928/comments"},"review_comments":{"href":"https://api.github.com/repos/ReactiveX/RxJava/pulls/2928/comments"},"review_comment":{"href":"https://api.github.com/repos/ReactiveX/RxJava/pulls/comments{/number}"},"commits":{"href":"https://api.github.com/repos/ReactiveX/RxJava/pulls/2928/commits"},"statuses":{"href":"https://api.github.com/repos/ReactiveX/RxJava/statuses/8a6d506bb199d439d72a9b884cdda624d178a4db"}}}}