Create IterablePublisher to convert an Iterable to a Publisher (#5030)

* Implement the functionality to convert an Iterable to a Publisher

* Use SimplePublisher in IterablePublisher

* Remove unncessary lock

.changes/next-release/feature-AWSSDKforJavav2-de94de6.json

@@ -0,0 +1,6 @@
+{
+    "type": "feature",
+    "category": "AWS SDK for Java v2",
+    "contributor": "",
+    "description": "Support creating an `SdkPublisher` from an `Iterable`"
+}

core/sdk-core/src/main/java/software/amazon/awssdk/core/async/SdkPublisher.java

@@ -30,6 +30,7 @@
 import software.amazon.awssdk.utils.async.EventListeningSubscriber;
 import software.amazon.awssdk.utils.async.FilteringSubscriber;
 import software.amazon.awssdk.utils.async.FlatteningSubscriber;
+import software.amazon.awssdk.utils.async.IterablePublisher;
 import software.amazon.awssdk.utils.async.LimitingSubscriber;
 import software.amazon.awssdk.utils.async.SequentialSubscriber;
 import software.amazon.awssdk.utils.internal.MappingSubscriber;
@@ -51,6 +52,17 @@ static <T> SdkPublisher<T> adapt(Publisher<T> toAdapt) {
         return toAdapt::subscribe;
     }
 
+    /**
+     * Create an {@link SdkPublisher} from an {@link Iterable}.
+     *
+     * @param iterable {@link Iterable} to adapt.
+     * @param <T> Type of object being published.
+     * @return SdkPublisher
+     */
+    static <T> SdkPublisher<T> fromIterable(Iterable<T> iterable) {
+        return adapt(new IterablePublisher<>(iterable));
+    }
+
     /**
      * Filters published events to just those that are instances of the given class. This changes the type of
      * publisher to the type specified in the {@link Class}.

utils/pom.xml

@@ -120,6 +120,11 @@
             <scope>test</scope>
             <version>${slf4j.version}</version>
         </dependency>
+        <dependency>
+            <groupId>commons-lang</groupId>
+            <artifactId>commons-lang</artifactId>
+            <scope>test</scope>
+        </dependency>
     </dependencies>
 
     <build>

utils/src/main/java/software/amazon/awssdk/utils/async/IterablePublisher.java

@@ -0,0 +1,67 @@
+/*
+ * Copyright Amazon.com, Inc. 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.
+ * A copy of the License is located at
+ *
+ *  http://aws.amazon.com/apache2.0
+ *
+ * or in the "license" file accompanying this file. This file 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 software.amazon.awssdk.utils.async;
+
+import java.util.Iterator;
+import org.reactivestreams.Publisher;
+import org.reactivestreams.Subscriber;
+import software.amazon.awssdk.annotations.SdkProtectedApi;
+import software.amazon.awssdk.utils.Validate;
+
+@SdkProtectedApi
+public class IterablePublisher<T> implements Publisher<T> {
+    private final Iterable<T> iterable;
+
+    public IterablePublisher(Iterable<T> iterable) {
+        this.iterable = Validate.paramNotNull(iterable, "iterable");
+    }
+
+    @Override
+    public void subscribe(Subscriber<? super T> subscriber) {
+        Iterator<T> iterator = iterable.iterator();
+        SimplePublisher<T> publisher = new SimplePublisher<>();
+
+        // Prime the simple publisher with 1 event. More will be sent as these complete.
+        sendEvent(iterator, publisher);
+
+        publisher.subscribe(subscriber);
+    }
+
+    private void sendEvent(Iterator<T> iterator, SimplePublisher<T> publisher) {
+        try {
+            if (!iterator.hasNext()) {
+                publisher.complete();
+                return;
+            }
+
+            T next = iterator.next();
+            if (next == null) {
+                publisher.error(new IllegalArgumentException("Iterable returned null"));
+                return;
+            }
+
+            publisher.send(next).whenComplete((v, t) -> {
+                if (t != null) {
+                    publisher.error(t);
+                } else {
+                    sendEvent(iterator, publisher);
+                }
+            });
+        } catch (Throwable e) {
+            publisher.error(e);
+        }
+    }
+}

utils/src/main/java/software/amazon/awssdk/utils/async/SimplePublisher.java

@@ -278,7 +278,7 @@ private void doProcessQueue() {
 
                         OnErrorQueueEntry<T> onErrorEntry = (OnErrorQueueEntry<T>) entry;
                         failureMessage.trySet(() -> new IllegalStateException("onError() was already invoked.",
-                                                                             onErrorEntry.failure));
+                                                                              onErrorEntry.failure));
                         log.trace(() -> "Calling onError() with " + onErrorEntry.failure, onErrorEntry.failure);
                         subscriber.onError(onErrorEntry.failure);
                         break;

utils/src/test/java/software/amazon/awssdk/utils/async/IterablePublisherTckTest.java

@@ -0,0 +1,56 @@
+/*
+ * Copyright Amazon.com, Inc. 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.
+ * A copy of the License is located at
+ *
+ *  http://aws.amazon.com/apache2.0
+ *
+ * or in the "license" file accompanying this file. This file 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 software.amazon.awssdk.utils.async;
+
+import java.util.Iterator;
+import org.reactivestreams.Publisher;
+import org.reactivestreams.tck.PublisherVerification;
+import org.reactivestreams.tck.TestEnvironment;
+
+public class IterablePublisherTckTest extends PublisherVerification<Long>  {
+
+
+    public IterablePublisherTckTest() {
+        super(new TestEnvironment());
+    }
+
+    @Override
+    public Publisher<Long> createPublisher(long elements) {
+        Iterable<Long> iterable = () -> new Iterator<Long>() {
+            private long count;
+            @Override
+            public boolean hasNext() {
+                if (count == elements) {
+                    return false;
+                }
+
+                return true;
+            }
+
+            @Override
+            public Long next() {
+                count++;
+                return count;
+            }
+        };
+        return new IterablePublisher<>(iterable);
+    }
+
+    @Override
+    public Publisher<Long> createFailedPublisher() {
+        return null;
+    }
+}

utils/src/test/java/software/amazon/awssdk/utils/async/IterablePublisherTest.java

@@ -0,0 +1,105 @@
+/*
+ * Copyright Amazon.com, Inc. 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.
+ * A copy of the License is located at
+ *
+ *  http://aws.amazon.com/apache2.0
+ *
+ * or in the "license" file accompanying this file. This file 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 software.amazon.awssdk.utils.async;
+
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+import java.util.stream.Collectors;
+import java.util.stream.IntStream;
+import org.apache.commons.lang.RandomStringUtils;
+import org.junit.jupiter.api.Test;
+import org.reactivestreams.Subscriber;
+import org.reactivestreams.Subscription;
+
+public class IterablePublisherTest {
+
+    @Test
+    void nullIterable_throwException() {
+        assertThatThrownBy(() -> new IterablePublisher<>(null)).isInstanceOf(NullPointerException.class);
+    }
+
+    @Test
+    void emptyIterable_shouldComplete() {
+        TestSubscriber testSubscriber = new TestSubscriber();
+        IterablePublisher<String> iterablePublisher = new IterablePublisher<>(new ArrayList<>());
+        iterablePublisher.subscribe(testSubscriber);
+        assertThat(testSubscriber.onCompleteInvoked).isTrue();
+        assertThat(testSubscriber.onNextInvoked).isFalse();
+        assertThat(testSubscriber.onErrorInvoked).isFalse();
+    }
+
+    @Test
+    void iterableReturnNull_shouldInvokeOnError() {
+        TestSubscriber testSubscriber = new TestSubscriber();
+        IterablePublisher<String> iterablePublisher = new IterablePublisher<>(Arrays.asList("foo", null));
+        iterablePublisher.subscribe(testSubscriber);
+        assertThat(testSubscriber.onCompleteInvoked).isFalse();
+        assertThat(testSubscriber.results).contains("foo");
+        assertThat(testSubscriber.onErrorInvoked).isTrue();
+        assertThat(testSubscriber.throwable).isInstanceOf(IllegalArgumentException.class).hasMessageContaining("returned null");
+    }
+
+    @Test
+    void happyCase_shouldSendAllEvents() {
+        TestSubscriber testSubscriber = new TestSubscriber();
+        List<String> strings = IntStream.range(0, 100).mapToObj(i -> RandomStringUtils.random(i)).collect(Collectors.toList());
+
+        IterablePublisher<String> iterablePublisher = new IterablePublisher<>(strings);
+        iterablePublisher.subscribe(testSubscriber);
+        assertThat(testSubscriber.onCompleteInvoked).isTrue();
+        assertThat(testSubscriber.results).hasSameElementsAs(strings);
+        assertThat(testSubscriber.onErrorInvoked).isFalse();
+    }
+
+    private static class TestSubscriber implements Subscriber<String> {
+        private Subscription subscription;
+        private List<String> results = new ArrayList<>();
+        private boolean onNextInvoked;
+        private boolean onErrorInvoked;
+        private boolean onCompleteInvoked;
+        private Throwable throwable;
+
+        @Override
+        public void onSubscribe(Subscription s) {
+            this.subscription = s;
+            s.request(1);
+        }
+
+        @Override
+        public void onNext(String s) {
+            onNextInvoked = true;
+            results.add(s);
+            subscription.request(1);
+        }
+
+        @Override
+        public void onError(Throwable t) {
+            onErrorInvoked = true;
+            throwable = t;
+        }
+
+        @Override
+        public void onComplete() {
+            onCompleteInvoked = true;
+
+        }
+    }
+}