[Reactive] Reimplement Concat with varargs

common/reactive/src/main/java/io/helidon/common/reactive/Multi.java

@@ -16,7 +16,6 @@
 package io.helidon.common.reactive;
 
 import java.util.ArrayList;
-import java.util.Arrays;
 import java.util.Collection;
 import java.util.List;
 import java.util.Objects;
@@ -64,25 +63,21 @@ static <T> Multi<T> concat(Flow.Publisher<T> firstMulti, Flow.Publisher<T> secon
     }
 
     /**
-     * Concat streams to one.
-     *
-     * @param firstMulti  first stream
-     * @param secondMulti second stream
+     * Concatenates an array of source {@link Flow.Publisher}s by relaying items
+     * in order, non-overlappingly, one after the other finishes.
      * @param publishers  more publishers to concat
      * @param <T>         item type
      * @return Multi
      */
     @SafeVarargs
     @SuppressWarnings("varargs")
-    static <T> Multi<T> concat(Flow.Publisher<T> firstMulti, Flow.Publisher<T> secondMulti, Flow.Publisher<T>... publishers) {
+    static <T> Multi<T> concatArray(Flow.Publisher<T>... publishers) {
         if (publishers.length == 0) {
-            return concat(firstMulti, secondMulti);
+            return empty();
         } else if (publishers.length == 1) {
-            return concat(concat(firstMulti, secondMulti), publishers[0]);
-        } else {
-            return concat(concat(firstMulti, secondMulti), publishers[0],
-                    Arrays.copyOfRange(publishers, 1, publishers.length));
+            return Multi.from(publishers[0]);
         }
+        return new MultiConcatArray<>(publishers);
     }
 
     /**

common/reactive/src/main/java/io/helidon/common/reactive/MultiConcatArray.java

@@ -0,0 +1,107 @@
+/*
+ * Copyright (c)  2020 Oracle and/or its affiliates. All rights reserved.
+ *
+ * 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 io.helidon.common.reactive;
+
+import java.util.concurrent.Flow;
+import java.util.concurrent.atomic.AtomicInteger;
+
+/**
+ * Relay items in order from subsequent Flow.Publishers as a single Multi source.
+ */
+final class MultiConcatArray<T> implements Multi<T> {
+
+    private final Flow.Publisher<T>[] sources;
+
+    MultiConcatArray(Flow.Publisher<T>[] sources) {
+        this.sources = sources;
+    }
+
+    @Override
+    public void subscribe(Flow.Subscriber<? super T> subscriber) {
+        ConcatArraySubscriber<T> parent = new ConcatArraySubscriber<>(subscriber, sources);
+        subscriber.onSubscribe(parent);
+        parent.nextSource();
+    }
+
+    static final class ConcatArraySubscriber<T> extends SubscriptionArbiter
+    implements Flow.Subscriber<T> {
+
+        private final Flow.Subscriber<? super T> downstream;
+
+        private final Flow.Publisher<T>[] sources;
+
+        private final AtomicInteger wip;
+
+        private int index;
+
+        private long produced;
+
+        ConcatArraySubscriber(Flow.Subscriber<? super T> downstream, Flow.Publisher<T>[] sources) {
+            this.downstream = downstream;
+            this.sources = sources;
+            this.wip = new AtomicInteger();
+        }
+
+        @Override
+        public void onSubscribe(Flow.Subscription subscription) {
+            super.setSubscription(subscription);
+        }
+
+        @Override
+        public void onNext(T item) {
+            produced++;
+            downstream.onNext(item);
+        }
+
+        @Override
+        public void onError(Throwable throwable) {
+            downstream.onError(throwable);
+        }
+
+        @Override
+        public void onComplete() {
+            long produced = this.produced;
+            if (produced != 0L) {
+                this.produced = 0L;
+                super.produced(produced);
+            }
+            nextSource();
+        }
+
+        public void nextSource() {
+            if (wip.getAndIncrement() == 0) {
+                do {
+                    if (index == sources.length) {
+                        downstream.onComplete();
+                    } else {
+                        sources[index++].subscribe(this);
+                    }
+                } while (wip.decrementAndGet() != 0);
+            }
+        }
+
+        @Override
+        public void request(long n) {
+            if (n <= 0) {
+                downstream.onError(new IllegalArgumentException("Rule §3.9 violated: non-positive requests are forbidden"));
+            } else {
+                super.request(n);
+            }
+        }
+    }
+}

common/reactive/src/test/java/io/helidon/common/reactive/MultiConcatArrayTck1Test.java

@@ -0,0 +1,47 @@
+/*
+ * Copyright (c)  2020 Oracle and/or its affiliates. All rights reserved.
+ *
+ * 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 io.helidon.common.reactive;
+
+import org.reactivestreams.tck.TestEnvironment;
+import org.reactivestreams.tck.flow.FlowPublisherVerification;
+import org.testng.annotations.Test;
+
+import java.util.Arrays;
+import java.util.concurrent.Flow;
+
+@Test
+public class MultiConcatArrayTck1Test extends FlowPublisherVerification<Integer> {
+
+    public MultiConcatArrayTck1Test() {
+        super(new TestEnvironment(200));
+    }
+
+    @Override
+    public Flow.Publisher<Integer> createFlowPublisher(long l) {
+        return Multi.concatArray(Multi.range(0, (int) l / 2), Multi.range((int)l / 2, (int) (l - l / 2)));
+    }
+
+    @Override
+    public Flow.Publisher<Integer> createFailedFlowPublisher() {
+        return null;
+    }
+
+    @Override
+    public long maxElementsFromPublisher() {
+        return 10;
+    }
+}

common/reactive/src/test/java/io/helidon/common/reactive/MultiConcatArrayTck2Test.java

@@ -0,0 +1,52 @@
+/*
+ * Copyright (c)  2020 Oracle and/or its affiliates. All rights reserved.
+ *
+ * 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 io.helidon.common.reactive;
+
+import org.reactivestreams.tck.TestEnvironment;
+import org.reactivestreams.tck.flow.FlowPublisherVerification;
+import org.testng.annotations.Test;
+
+import java.util.Arrays;
+import java.util.concurrent.Flow;
+
+@Test
+public class MultiConcatArrayTck2Test extends FlowPublisherVerification<Integer> {
+
+    public MultiConcatArrayTck2Test() {
+        super(new TestEnvironment(200));
+    }
+
+    @Override
+    public Flow.Publisher<Integer> createFlowPublisher(long l) {
+        @SuppressWarnings("unchecked")
+        Multi<Integer>[] sources = new Multi[(int)l];
+        for (int i = 0; i < l; i++) {
+            sources[i] = Multi.singleton(i);
+        }
+        return Multi.concatArray(sources);
+    }
+
+    @Override
+    public Flow.Publisher<Integer> createFailedFlowPublisher() {
+        return null;
+    }
+
+    @Override
+    public long maxElementsFromPublisher() {
+        return 10;
+    }
+}

common/reactive/src/test/java/io/helidon/common/reactive/MultiConcatArrayTest.java

@@ -0,0 +1,54 @@
+/*
+ * Copyright (c) 2020 Oracle and/or its affiliates. All rights reserved.
+ *
+ * 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 io.helidon.common.reactive;
+
+import org.junit.jupiter.api.Test;
+
+import java.io.IOException;
+import java.util.Arrays;
+
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.core.Is.is;
+import static org.hamcrest.core.IsInstanceOf.instanceOf;
+
+public class MultiConcatArrayTest {
+
+    @Test
+    public void errors() {
+        TestSubscriber<Object> ts = new TestSubscriber<>(Long.MAX_VALUE);
+
+        Multi.concatArray(Multi.singleton(1), Multi.error(new IOException()), Multi.singleton(2))
+        .subscribe(ts);
+
+        ts.assertFailure(IOException.class, 1);
+    }
+
+    @Test
+    public void millionSources() {
+        @SuppressWarnings("unchecked")
+        Multi<Integer>[] sources = new Multi[1_000_000];
+        Arrays.fill(sources, Multi.singleton(1));
+
+        TestSubscriber<Object> ts = new TestSubscriber<>(Long.MAX_VALUE);
+
+        Multi.concatArray(sources)
+                .subscribe(ts);
+
+        ts.assertItemCount(1_000_000)
+                .assertComplete();
+    }
+
+}

common/reactive/src/test/java/io/helidon/common/reactive/MultiTest.java

@@ -530,7 +530,7 @@ void testConcatVarargs() {
 
 
         assertThat(Multi
-                .concat(Multi.from(TEST_DATA_1),
+                .concatArray(Multi.from(TEST_DATA_1),
                         Multi.just(TEST_DATA_2),
                         Multi.just(TEST_DATA_3)
                 )
@@ -542,7 +542,7 @@ void testConcatVarargs() {
         )))));
 
         assertThat(Multi
-                .concat(Multi.from(TEST_DATA_1),
+                .concatArray(Multi.from(TEST_DATA_1),
                         Multi.just(TEST_DATA_2),
                         Multi.just(TEST_DATA_3),
                         Multi.just(TEST_DATA_4)
@@ -557,7 +557,7 @@ void testConcatVarargs() {
 
 
         assertThat(Multi
-                        .concat(Multi.from(TEST_DATA_1),
+                        .concatArray(Multi.from(TEST_DATA_1),
                                 Multi.just(TEST_DATA_2),
                                 Multi.just(TEST_DATA_3),
                                 Multi.just(TEST_DATA_4),
@@ -575,7 +575,7 @@ void testConcatVarargs() {
 
 
         assertThat(Multi
-                        .concat(Multi.from(TEST_DATA_1),
+                        .concatArray(Multi.from(TEST_DATA_1),
                                 Multi.just(TEST_DATA_2),
                                 Multi.just(TEST_DATA_3),
                                 Multi.just(TEST_DATA_4),