ParallelFlowable.as()

Author: ZacSweers

src/main/java/io/reactivex/parallel/ParallelFlowable.java

@@ -122,6 +122,29 @@ public static <T> ParallelFlowable<T> from(@NonNull Publisher<? extends T> sourc
         return RxJavaPlugins.onAssembly(new ParallelFromPublisher<T>(source, parallelism, prefetch));
     }
 
+    /**
+     * Calls the specified converter function during assembly time and returns its resulting value.
+     * <p>
+     * This allows fluent conversion to any other type.
+     *
+     * @param <R> the resulting object type
+     * @param converter the function that receives the current ParallelFlowable instance and returns a value
+     * @return the converted value
+     * @throws NullPointerException if converter is null
+     * @since 2.1.7 - experimental
+     */
+    @Experimental
+    @CheckReturnValue
+    @NonNull
+    public final <R> R as(@NonNull ParallelFlowableConverter<T, R> converter) {
+        try {
+            return ObjectHelper.requireNonNull(converter, "converter is null").apply(this);
+        } catch (Throwable ex) {
+            Exceptions.throwIfFatal(ex);
+            throw ExceptionHelper.wrapOrThrow(ex);
+        }
+    }
+
     /**
      * Maps the source values on each 'rail' to another value.
      * <p>

src/main/java/io/reactivex/parallel/ParallelFlowableConverter.java

@@ -0,0 +1,37 @@
+/**
+ * Copyright (c) 2016-present, RxJava Contributors.
+ *
+ * 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.reactivex.parallel;
+
+import io.reactivex.annotations.*;
+
+/**
+ * Convenience interface and callback used by the {@link ParallelFlowable#as} operator to turn a ParallelFlowable into
+ * another value fluently.
+ *
+ * @param <T> the upstream type
+ * @param <R> the output type
+ * @since 2.1.7 - experimental
+ */
+@Experimental
+public interface ParallelFlowableConverter<T, R> {
+    /**
+     * Applies a function to the upstream ParallelFlowable and returns a converted value of type {@code R}.
+     *
+     * @param upstream the upstream ParallelFlowable instance
+     * @return the converted value
+     * @throws Exception on error
+     */
+    @NonNull
+    R apply(@NonNull ParallelFlowable<T> upstream) throws Exception;
+}

src/test/java/io/reactivex/parallel/ParallelFlowableTest.java

@@ -1100,6 +1100,20 @@ public Flowable<Integer> apply(ParallelFlowable<Integer> pf) throws Exception {
         .assertResult(1, 2, 3, 4, 5);
     }
 
+    @Test
+    public void as() {
+        Flowable.range(1, 5)
+        .parallel()
+        .as(new ParallelFlowableConverter<Integer, Flowable<Integer>>() {
+            @Override
+            public Flowable<Integer> apply(ParallelFlowable<Integer> pf) throws Exception {
+                return pf.sequential();
+            }
+        })
+        .test()
+        .assertResult(1, 2, 3, 4, 5);
+    }
+
     @Test(expected = TestException.class)
     public void toThrows() {
         Flowable.range(1, 5)
@@ -1112,6 +1126,18 @@ public Flowable<Integer> apply(ParallelFlowable<Integer> pf) throws Exception {
         });
     }
 
+    @Test(expected = TestException.class)
+    public void asThrows() {
+        Flowable.range(1, 5)
+        .parallel()
+        .as(new ParallelFlowableConverter<Integer, Flowable<Integer>>() {
+            @Override
+            public Flowable<Integer> apply(ParallelFlowable<Integer> pf) throws Exception {
+                throw new TestException();
+            }
+        });
+    }
+
     @Test
     public void compose() {
         Flowable.range(1, 5)