Merge pull request #1412 from benjchristensen/backpressure-9

Backpressure & 0.20.0-RC1

Author: benjchristensen

build.gradle

@@ -106,6 +106,7 @@ subprojects {
         maxHeapSize = "512m"
 		jvmArgs '-XX:+UnlockCommercialFeatures'
 		jvmArgs '-XX:+FlightRecorder'
+		jvmArgs '-XX:AutoBoxCacheMax=1000000'
 
         if (project.hasProperty('jmh')) {
             args(jmh.split(' '))

gradle.properties

@@ -1 +1 @@
-version=0.19.5-SNAPSHOT
+version=0.20.0-RC1-SNAPSHOT

rxjava-core/src/main/java/rx/Observable.java

@@ -20,6 +20,7 @@
 import rx.exceptions.*;
 import rx.functions.*;
 import rx.internal.operators.*;
+import rx.internal.util.ScalarSynchronousObservable;
 import rx.observables.*;
 import rx.observers.SafeSubscriber;
 import rx.plugins.*;
@@ -1154,7 +1155,7 @@ public final static <T> Observable<T> from(Iterable<? extends T> iterable) {
      * @see <a href="http://msdn.microsoft.com/en-us/library/hh212140.aspx">MSDN: Observable.ToObservable</a>
      */
     public final static <T> Observable<T> from(Iterable<? extends T> iterable, Scheduler scheduler) {
-        return create(new OnSubscribeFromIterable<T>(iterable)).subscribeOn(scheduler);
+        return from(iterable).subscribeOn(scheduler);
     }
 
     /**
@@ -1543,17 +1544,7 @@ public final static Observable<Long> interval(long interval, TimeUnit unit, Sche
      * @see <a href="https://github.com/Netflix/RxJava/wiki/Creating-Observables#just">RxJava Wiki: just()</a>
      */
     public final static <T> Observable<T> just(final T value) {
-        return Observable.create(new OnSubscribe<T>() {
-
-            @Override
-            public void call(Subscriber<? super T> s) {
-                if (!s.isUnsubscribed()) {
-                    s.onNext(value);
-                    s.onCompleted();
-                }
-            }
-
-        });
+        return ScalarSynchronousObservable.create(value);
     }
 
     /**
@@ -2471,6 +2462,9 @@ public final static Observable<Integer> range(int start, int count) {
         if (start > Integer.MAX_VALUE - count + 1) {
             throw new IllegalArgumentException("start + count can not exceed Integer.MAX_VALUE");
         }
+        if(count == 1) {
+            return Observable.just(start);
+        }
         return Observable.create(new OnSubscribeRange(start, start + (count - 1)));
     }
 
@@ -4914,6 +4908,14 @@ public final Boolean call(T t) {
         }).cast(klass);
     }
 
+    public final Observable<T> onBackpressureBuffer() {
+        return lift(new OperatorOnBackpressureBuffer<T>());
+    }
+    
+    public final Observable<T> onBackpressureDrop() {
+        return lift(new OperatorOnBackpressureDrop<T>());
+    }
+    
     /**
      * Instructs an Observable to pass control to another Observable rather than invoking
      * {@link Observer#onError onError} if it encounters an error.

rxjava-core/src/main/java/rx/Producer.java

@@ -0,0 +1,22 @@
+/**
+ * Copyright 2014 Netflix, Inc.
+ * 
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ * 
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * 
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package rx;
+
+public interface Producer {
+
+    public void request(int n);
+
+}

rxjava-core/src/main/java/rx/Subscriber.java

@@ -34,27 +34,48 @@
 public abstract class Subscriber<T> implements Observer<T>, Subscription {
 
     private final SubscriptionList cs;
+    private final Subscriber<?> op;
+    /* protected by `this` */
+    private Producer p;
+    /* protected by `this` */
+    private int requested = Integer.MIN_VALUE; // default to not set
 
     @Deprecated
     protected Subscriber(CompositeSubscription cs) {
+        this.op = null;
         this.cs = new SubscriptionList();
         add(cs);
     }
 
     protected Subscriber() {
+        this.op = null;
         this.cs = new SubscriptionList();
     }
 
+    protected Subscriber(int bufferRequest) {
+        this.op = null;
+        this.cs = new SubscriptionList();
+        request(bufferRequest);
+    }
+
     protected Subscriber(Subscriber<?> op) {
+        this.op = op;
         this.cs = op.cs;
     }
 
+    protected Subscriber(Subscriber<?> op, int bufferRequest) {
+        this.op = op;
+        this.cs = op.cs;
+        request(bufferRequest);
+    }
+
     /**
      * Adds a {@link Subscription} to this Subscriber's list of subscriptions if this list is not marked as
      * unsubscribed. If the list <em>is</em> marked as unsubscribed, {@code add} will indicate this by
      * explicitly unsubscribing the new {@code Subscription} as well.
      *
-     * @param s the {@code Subscription} to add
+     * @param s
+     *            the {@code Subscription} to add
      */
     public final void add(Subscription s) {
         cs.add(s);
@@ -73,4 +94,51 @@ public final void unsubscribe() {
     public final boolean isUnsubscribed() {
         return cs.isUnsubscribed();
     }
+
+    public final void request(int n) {
+        Producer shouldRequest = null;
+        synchronized (this) {
+            if (p != null) {
+                shouldRequest = p;
+            } else {
+                requested = n;
+            }
+        }
+        // after releasing lock
+        if (shouldRequest != null) {
+            shouldRequest.request(n);
+        }
+    }
+
+    protected Producer onSetProducer(Producer producer) {
+        return producer;
+    }
+
+    public final void setProducer(Producer producer) {
+        producer = onSetProducer(producer);
+        int toRequest = requested;
+        boolean setProducer = false;
+        synchronized (this) {
+            p = producer;
+            if (op != null) {
+                // middle operator ... we pass thru unless a request has been made
+                if (toRequest == Integer.MIN_VALUE) {
+                    // we pass-thru to the next producer as nothing has been requested
+                    setProducer = true;
+                }
+
+            }
+        }
+        // do after releasing lock
+        if (setProducer) {
+            op.setProducer(p);
+        } else {
+            // we execute the request with whatever has been requested (or -1)
+            if (toRequest == Integer.MIN_VALUE) {
+                p.request(-1);
+            } else {
+                p.request(toRequest);
+            }
+        }
+    }
 }

rxjava-core/src/main/java/rx/exceptions/MissingBackpressureException.java

@@ -0,0 +1,29 @@
+/**
+ * Copyright 2014 Netflix, Inc.
+ * 
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ * 
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * 
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package rx.exceptions;
+
+public class MissingBackpressureException extends Exception {
+
+    private static final long serialVersionUID = 7250870679677032194L;
+
+    public MissingBackpressureException() {
+    }
+
+    public MissingBackpressureException(String message) {
+        super(message);
+    }
+
+}

rxjava-core/src/main/java/rx/internal/operators/OnSubscribeFromIterable.java

@@ -15,7 +15,12 @@
  */
 package rx.internal.operators;
 
+import java.util.Collection;
+import java.util.Iterator;
+import java.util.concurrent.atomic.AtomicIntegerFieldUpdater;
+
 import rx.Observable.OnSubscribe;
+import rx.Producer;
 import rx.Subscriber;
 
 /**
@@ -35,17 +40,71 @@ public OnSubscribeFromIterable(Iterable<? extends T> iterable) {
     }
 
     @Override
-    public void call(Subscriber<? super T> o) {
-        for (T i : is) {
-            if (o.isUnsubscribed()) {
-                return;
-            }
-            o.onNext(i);
+    public void call(final Subscriber<? super T> o) {
+        if (is == null) {
+            o.onCompleted();
+        }
+        final Iterator<? extends T> it = is.iterator();
+        o.setProducer(new IterableProducer<T>(o, it));
+    }
+
+    private static final class IterableProducer<T> implements Producer {
+        private final Subscriber<? super T> o;
+        private final Iterator<? extends T> it;
+
+        private volatile int requested = 0;
+        @SuppressWarnings("rawtypes")
+        private static final AtomicIntegerFieldUpdater<IterableProducer> REQUESTED_UPDATER = AtomicIntegerFieldUpdater.newUpdater(IterableProducer.class, "requested");
+
+        private IterableProducer(Subscriber<? super T> o, Iterator<? extends T> it) {
+            this.o = o;
+            this.it = it;
         }
-        if (o.isUnsubscribed()) {
-            return;
+
+        @Override
+        public void request(int n) {
+            if (n < 0) {
+                // fast-path without backpressure
+                while (it.hasNext()) {
+                    if (o.isUnsubscribed()) {
+                        return;
+                    }
+                    o.onNext(it.next());
+                }
+                o.onCompleted();
+            } else {
+                // backpressure is requested
+                int _c = REQUESTED_UPDATER.getAndAdd(this, n);
+                if (_c == 0) {
+                    while (true) {
+                        /*
+                         * This complicated logic is done to avoid touching the volatile `requested` value
+                         * during the loop itself. If it is touched during the loop the performance is impacted significantly.
+                         */
+                        int r = requested;
+                        int numToEmit = r;
+                        while (it.hasNext() && --numToEmit >= 0) {
+                            if (o.isUnsubscribed()) {
+                                return;
+                            }
+                            o.onNext(it.next());
+
+                        }
+
+                        if (!it.hasNext()) {
+                            o.onCompleted();
+                            return;
+                        }
+                        if (REQUESTED_UPDATER.addAndGet(this, -r) == 0) {
+                            // we're done emitting the number requested so return
+                            return;
+                        }
+
+                    }
+                }
+            }
+
         }
-        o.onCompleted();
     }
 
 }