1.x: Add maxConcurrent parameter to concatMapEager

* Add maxConcurrent parameter to concatMapEager

* Improve ConcatMapEager performance and add unit test

* Add test case and cleanup whitespace

Author: ADI1133

src/main/java/rx/Observable.java

@@ -5353,7 +5353,41 @@ public final <R> Observable<R> concatMapEager(Func1<? super T, ? extends Observa
         if (capacityHint < 1) {
             throw new IllegalArgumentException("capacityHint > 0 required but it was " + capacityHint);
         }
-        return lift(new OperatorEagerConcatMap<T, R>(mapper, capacityHint));
+        return lift(new OperatorEagerConcatMap<T, R>(mapper, capacityHint, Integer.MAX_VALUE));
+    }
+
+    /**
+     * Maps a sequence of values into Observables and concatenates these Observables eagerly into a single
+     * Observable.
+     * <p>
+     * Eager concatenation means that once a subscriber subscribes, this operator subscribes to all of the
+     * source Observables. The operator buffers the values emitted by these Observables and then drains them in
+     * order, each one after the previous one completes.
+     * <dl>
+     *  <dt><b>Backpressure:</b></dt>
+     *  <dd>Backpressure is honored towards the downstream, however, due to the eagerness requirement, sources
+     *      are subscribed to in unbounded mode and their values are queued up in an unbounded buffer.</dd>
+     *  <dt><b>Scheduler:</b></dt>
+     *  <dd>This method does not operate by default on a particular {@link Scheduler}.</dd>
+     * </dl>
+     * @param <R> the value type
+     * @param mapper the function that maps a sequence of values into a sequence of Observables that will be
+     *               eagerly concatenated
+     * @param capacityHint hints about the number of expected source sequence values
+     * @param maxConcurrent the maximum number of concurrent subscribed observables
+     * @return
+     * @warn javadoc fails to describe the return value
+     * @since (if this graduates from Experimental/Beta to supported, replace this parenthetical with the release number)
+     */
+    @Experimental
+    public final <R> Observable<R> concatMapEager(Func1<? super T, ? extends Observable<? extends R>> mapper, int capacityHint, int maxConcurrent) {
+        if (capacityHint < 1) {
+            throw new IllegalArgumentException("capacityHint > 0 required but it was " + capacityHint);
+        }
+        if (maxConcurrent < 1) {
+            throw new IllegalArgumentException("maxConcurrent > 0 required but it was " + capacityHint);
+        }
+        return lift(new OperatorEagerConcatMap<T, R>(mapper, capacityHint, maxConcurrent));
     }
 
     /**

src/main/java/rx/internal/operators/OperatorEagerConcatMap.java

@@ -31,14 +31,16 @@
 public final class OperatorEagerConcatMap<T, R> implements Operator<R, T> {
     final Func1<? super T, ? extends Observable<? extends R>> mapper;
     final int bufferSize;
-    public OperatorEagerConcatMap(Func1<? super T, ? extends Observable<? extends R>> mapper, int bufferSize) {
+    private final int maxConcurrent;
+    public OperatorEagerConcatMap(Func1<? super T, ? extends Observable<? extends R>> mapper, int bufferSize, int maxConcurrent) {
         this.mapper = mapper;
         this.bufferSize = bufferSize;
+        this.maxConcurrent = maxConcurrent;
     }
 
     @Override
     public Subscriber<? super T> call(Subscriber<? super R> t) {
-        EagerOuterSubscriber<T, R> outer = new EagerOuterSubscriber<T, R>(mapper, bufferSize, t);
+        EagerOuterSubscriber<T, R> outer = new EagerOuterSubscriber<T, R>(mapper, bufferSize, maxConcurrent, t);
         outer.init();
         return outer;
     }
@@ -82,12 +84,13 @@ static final class EagerOuterSubscriber<T, R> extends Subscriber<T> {
         private EagerOuterProducer sharedProducer;
 
         public EagerOuterSubscriber(Func1<? super T, ? extends Observable<? extends R>> mapper, int bufferSize,
-                Subscriber<? super R> actual) {
+                int maxConcurrent, Subscriber<? super R> actual) {
             this.mapper = mapper;
             this.bufferSize = bufferSize;
             this.actual = actual;
             this.subscribers = new LinkedList<EagerInnerSubscriber<R>>();
             this.wip = new AtomicInteger();
+            request(maxConcurrent == Integer.MAX_VALUE ? Long.MAX_VALUE : maxConcurrent);
         }
 
         void init() {
@@ -223,6 +226,7 @@ void drain() {
                                 }
                                 innerSubscriber.unsubscribe();
                                 innerDone = true;
+                                request(1);
                                 break;
                             }
                         }

src/test/java/rx/internal/operators/OperatorEagerConcatMapTest.java

@@ -16,6 +16,8 @@
 
 package rx.internal.operators;
 
+import java.util.ArrayList;
+import java.util.List;
 import java.util.concurrent.atomic.*;
 
 import org.junit.*;
@@ -302,11 +304,16 @@ public Observable<Integer> call(Integer t) {
         ts.assertNotCompleted();
         ts.assertError(TestException.class);
     }
-    
+
     @Test(expected = IllegalArgumentException.class)
     public void testInvalidCapacityHint() {
         Observable.just(1).concatMapEager(toJust, 0);
     }
+
+    @Test(expected = IllegalArgumentException.class)
+    public void testInvalidMaxConcurrent() {
+        Observable.just(1).concatMapEager(toJust, RxRingBuffer.SIZE, 0);
+    }
 
     @Test
     public void testBackpressure() {
@@ -397,17 +404,38 @@ public void call(Integer t) {
 
     @Test
     public void testInnerNull() {
-        TestSubscriber<Object> ts = TestSubscriber.create();
-        
         Observable.just(1).concatMapEager(new Func1<Integer, Observable<Integer>>() {
             @Override
             public Observable<Integer> call(Integer t) {
                 return Observable.just(null);
             }
         }).subscribe(ts);
-        
+
         ts.assertNoErrors();
         ts.assertCompleted();
         ts.assertValue(null);
     }
+
+
+    @Test
+    public void testMaxConcurrent5() {
+        final List<Long> requests = new ArrayList<Long>();
+        Observable.range(1, 100).doOnRequest(new Action1<Long>() {
+            @Override
+            public void call(Long reqCount) {
+                requests.add(reqCount);
+            }
+        }).concatMapEager(toJust, RxRingBuffer.SIZE, 5).subscribe(ts);
+
+        ts.assertNoErrors();
+        ts.assertValueCount(100);
+        ts.assertCompleted();
+
+        Assert.assertEquals(5, (long) requests.get(0));
+        Assert.assertEquals(1, (long) requests.get(1));
+        Assert.assertEquals(1, (long) requests.get(2));
+        Assert.assertEquals(1, (long) requests.get(3));
+        Assert.assertEquals(1, (long) requests.get(4));
+        Assert.assertEquals(1, (long) requests.get(5));
+    }
 }