2.x: add javadoc and unit test to the recently changed XTransformers

src/main/java/io/reactivex/CompletableTransformer.java

@@ -18,5 +18,12 @@
  * Completable fluently.
  */
 public interface CompletableTransformer {
-    CompletableSource apply(Completable completable) throws Exception;
+    /**
+     * Applies a function to the upstream Completable and returns a CompletableSource.
+     * @param upstream the upstream Completable instance
+     * @return the transformed CompletableSource instance
+     * @throws Exception in case the transformation throws, checked exceptions will be wrapped
+     * into a RuntimeException
+     */
+    CompletableSource apply(Completable upstream) throws Exception;
 }

src/main/java/io/reactivex/FlowableTransformer.java

@@ -22,5 +22,13 @@
  * @param <Downstream> the downstream value type
  */
 public interface FlowableTransformer<Upstream, Downstream> {
-    Publisher<? extends Downstream> apply(Flowable<Upstream> flowable) throws Exception;
+    /**
+     * Applies a function to the upstream Flowable and returns a Publisher with
+     * optionally different element type.
+     * @param upstream the upstream Flowable instance
+     * @return the transformed Publisher instance
+     * @throws Exception in case the transformation throws, checked exceptions will be wrapped
+     * into a RuntimeException
+     */
+    Publisher<Downstream> apply(Flowable<Upstream> upstream) throws Exception;
 }

src/main/java/io/reactivex/MaybeTransformer.java

@@ -20,5 +20,13 @@
  * @param <Downstream> the downstream value type
  */
 public interface MaybeTransformer<Upstream, Downstream> {
-    MaybeSource<Downstream> apply(Maybe<Upstream> maybe) throws Exception;
+    /**
+     * Applies a function to the upstream Maybe and returns a MaybeSource with
+     * optionally different element type.
+     * @param upstream the upstream Maybe instance
+     * @return the transformed MaybeSource instance
+     * @throws Exception in case the transformation throws, checked exceptions will be wrapped
+     * into a RuntimeException
+     */
+    MaybeSource<Downstream> apply(Maybe<Upstream> upstream) throws Exception;
 }

src/main/java/io/reactivex/ObservableTransformer.java

@@ -20,5 +20,13 @@
  * @param <Downstream> the downstream value type
  */
 public interface ObservableTransformer<Upstream, Downstream> {
+    /**
+     * Applies a function to the upstream Observable and returns an ObservableSource with
+     * optionally different element type.
+     * @param upstream the upstream Observable instance
+     * @return the transformed ObservableSource instance
+     * @throws Exception in case the transformation throws, checked exceptions will be wrapped
+     * into a RuntimeException
+     */
     ObservableSource<Downstream> apply(Observable<Upstream> upstream) throws Exception;
 }

src/main/java/io/reactivex/SingleTransformer.java

@@ -20,5 +20,13 @@
  * @param <Downstream> the downstream value type
  */
 public interface SingleTransformer<Upstream, Downstream> {
+    /**
+     * Applies a function to the upstream Single and returns a SingleSource with
+     * optionally different element type.
+     * @param upstream the upstream Single instance
+     * @return the transformed SingleSource instance
+     * @throws Exception in case the transformation throws, checked exceptions will be wrapped
+     * into a RuntimeException
+     */
     SingleSource<Downstream> apply(Single<Upstream> upstream) throws Exception;
 }

src/test/java/io/reactivex/TransformerTest.java

@@ -0,0 +1,181 @@
+/**
+ * Copyright 2016 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 io.reactivex;
+
+import static org.junit.Assert.*;
+
+import java.io.IOException;
+
+import org.junit.Test;
+import org.reactivestreams.Publisher;
+
+import io.reactivex.exceptions.TestException;
+
+public class TransformerTest {
+
+    @Test
+    public void flowableTransformerThrows() {
+        try {
+            Flowable.just(1).compose(new FlowableTransformer<Integer, Integer>() {
+                @Override
+                public Publisher<Integer> apply(Flowable<Integer> v) throws Exception {
+                    throw new TestException("Forced failure");
+                }
+            });
+            fail("Should have thrown!");
+        } catch (TestException ex) {
+            assertEquals("Forced failure", ex.getMessage());
+        }
+    }
+
+    @Test
+    public void flowableTransformerThrowsChecked() {
+        try {
+            Flowable.just(1).compose(new FlowableTransformer<Integer, Integer>() {
+                @Override
+                public Publisher<Integer> apply(Flowable<Integer> v) throws Exception {
+                    throw new IOException("Forced failure");
+                }
+            });
+            fail("Should have thrown!");
+        } catch (RuntimeException ex) {
+            assertTrue(ex.toString(), ex.getCause() instanceof IOException);
+            assertEquals("Forced failure", ex.getCause().getMessage());
+        }
+    }
+
+    @Test
+    public void observableTransformerThrows() {
+        try {
+            Observable.just(1).compose(new ObservableTransformer<Integer, Integer>() {
+                @Override
+                public Observable<Integer> apply(Observable<Integer> v) throws Exception {
+                    throw new TestException("Forced failure");
+                }
+            });
+            fail("Should have thrown!");
+        } catch (TestException ex) {
+            assertEquals("Forced failure", ex.getMessage());
+        }
+    }
+
+    @Test
+    public void observableTransformerThrowsChecked() {
+        try {
+            Observable.just(1).compose(new ObservableTransformer<Integer, Integer>() {
+                @Override
+                public Observable<Integer> apply(Observable<Integer> v) throws Exception {
+                    throw new IOException("Forced failure");
+                }
+            });
+            fail("Should have thrown!");
+        } catch (RuntimeException ex) {
+            assertTrue(ex.toString(), ex.getCause() instanceof IOException);
+            assertEquals("Forced failure", ex.getCause().getMessage());
+        }
+    }
+
+    @Test
+    public void singleTransformerThrows() {
+        try {
+            Single.just(1).compose(new SingleTransformer<Integer, Integer>() {
+                @Override
+                public Single<Integer> apply(Single<Integer> v) throws Exception {
+                    throw new TestException("Forced failure");
+                }
+            });
+            fail("Should have thrown!");
+        } catch (TestException ex) {
+            assertEquals("Forced failure", ex.getMessage());
+        }
+    }
+
+    @Test
+    public void singleTransformerThrowsChecked() {
+        try {
+            Single.just(1).compose(new SingleTransformer<Integer, Integer>() {
+                @Override
+                public Single<Integer> apply(Single<Integer> v) throws Exception {
+                    throw new IOException("Forced failure");
+                }
+            });
+            fail("Should have thrown!");
+        } catch (RuntimeException ex) {
+            assertTrue(ex.toString(), ex.getCause() instanceof IOException);
+            assertEquals("Forced failure", ex.getCause().getMessage());
+        }
+    }
+
+    @Test
+    public void maybeTransformerThrows() {
+        try {
+            Maybe.just(1).compose(new MaybeTransformer<Integer, Integer>() {
+                @Override
+                public Maybe<Integer> apply(Maybe<Integer> v) throws Exception {
+                    throw new TestException("Forced failure");
+                }
+            });
+            fail("Should have thrown!");
+        } catch (TestException ex) {
+            assertEquals("Forced failure", ex.getMessage());
+        }
+    }
+
+    @Test
+    public void maybeTransformerThrowsChecked() {
+        try {
+            Maybe.just(1).compose(new MaybeTransformer<Integer, Integer>() {
+                @Override
+                public Maybe<Integer> apply(Maybe<Integer> v) throws Exception {
+                    throw new IOException("Forced failure");
+                }
+            });
+            fail("Should have thrown!");
+        } catch (RuntimeException ex) {
+            assertTrue(ex.toString(), ex.getCause() instanceof IOException);
+            assertEquals("Forced failure", ex.getCause().getMessage());
+        }
+    }
+
+    @Test
+    public void completabeTransformerThrows() {
+        try {
+            Completable.complete().compose(new CompletableTransformer() {
+                @Override
+                public Completable apply(Completable v) throws Exception {
+                    throw new TestException("Forced failure");
+                }
+            });
+            fail("Should have thrown!");
+        } catch (TestException ex) {
+            assertEquals("Forced failure", ex.getMessage());
+        }
+    }
+
+    @Test
+    public void completabeTransformerThrowsChecked() {
+        try {
+            Completable.complete().compose(new CompletableTransformer() {
+                @Override
+                public Completable apply(Completable v) throws Exception {
+                    throw new IOException("Forced failure");
+                }
+            });
+            fail("Should have thrown!");
+        } catch (RuntimeException ex) {
+            assertTrue(ex.toString(), ex.getCause() instanceof IOException);
+            assertEquals("Forced failure", ex.getCause().getMessage());
+        }
+    }
+}

src/test/java/io/reactivex/flowable/FlowableCovarianceTest.java

@@ -92,9 +92,9 @@ public void accept(Movie v) {
                                 System.out.println(v);
                             }
                         })
-                        .compose(new FlowableTransformer<Movie, Object>() {
+                        .compose(new FlowableTransformer<Movie, Movie>() {
                             @Override
-                            public Publisher<? extends Object> apply(Flowable<Movie> m) {
+                            public Publisher<Movie> apply(Flowable<Movie> m) {
                                 return m.concatWith(Flowable.just(new ActionMovie()));
                             }
                         }
@@ -120,7 +120,7 @@ public void testCovarianceOfCompose() {
         Flowable<HorrorMovie> movie = Flowable.just(new HorrorMovie());
         Flowable<Movie> movie2 = movie.compose(new FlowableTransformer<HorrorMovie, Movie>() {
             @Override
-            public Publisher<? extends Movie> apply(Flowable<HorrorMovie> t) {
+            public Publisher<Movie> apply(Flowable<HorrorMovie> t) {
                 return Flowable.just(new Movie());
             }
         });
@@ -132,7 +132,7 @@ public void testCovarianceOfCompose2() {
         Flowable<Movie> movie = Flowable.<Movie> just(new HorrorMovie());
         Flowable<HorrorMovie> movie2 = movie.compose(new FlowableTransformer<Movie, HorrorMovie>() {
             @Override
-            public Publisher<? extends HorrorMovie> apply(Flowable<Movie> t) {
+            public Publisher<HorrorMovie> apply(Flowable<Movie> t) {
                 return Flowable.just(new HorrorMovie());
             }
         });
@@ -144,7 +144,7 @@ public void testCovarianceOfCompose3() {
         Flowable<Movie> movie = Flowable.<Movie>just(new HorrorMovie());
         Flowable<HorrorMovie> movie2 = movie.compose(new FlowableTransformer<Movie, HorrorMovie>() {
             @Override
-            public Publisher<? extends HorrorMovie> apply(Flowable<Movie> t) {
+            public Publisher<HorrorMovie> apply(Flowable<Movie> t) {
                 return Flowable.just(new HorrorMovie()).map(new Function<HorrorMovie, HorrorMovie>() {
                     @Override
                     public HorrorMovie apply(HorrorMovie v) {
@@ -162,7 +162,7 @@ public void testCovarianceOfCompose4() {
         Flowable<HorrorMovie> movie = Flowable.just(new HorrorMovie());
         Flowable<HorrorMovie> movie2 = movie.compose(new FlowableTransformer<HorrorMovie, HorrorMovie>() {
             @Override
-            public Publisher<? extends HorrorMovie> apply(Flowable<HorrorMovie> t1) {
+            public Publisher<HorrorMovie> apply(Flowable<HorrorMovie> t1) {
                 return t1.map(new Function<HorrorMovie, HorrorMovie>() {
                     @Override
                     public HorrorMovie apply(HorrorMovie v) {
@@ -211,7 +211,7 @@ public Flowable<Movie> apply(List<List<Movie>> listOfLists) {
 
     static FlowableTransformer<List<Movie>, Movie> deltaTransformer = new FlowableTransformer<List<Movie>, Movie>() {
         @Override
-        public Publisher<? extends Movie> apply(Flowable<List<Movie>> movieList) {
+        public Publisher<Movie> apply(Flowable<List<Movie>> movieList) {
             return movieList
                 .startWith(new ArrayList<Movie>())
                 .buffer(2, 1)