2.x: add most relevant ~100 operators' Reactive-Streams TCK tests (#4538)

Author: akarnokd

src/main/java/io/reactivex/internal/operators/flowable/FlowableUnsubscribeOn.java

@@ -17,8 +17,9 @@
 
 import org.reactivestreams.*;
 
-import io.reactivex.*;
+import io.reactivex.Scheduler;
 import io.reactivex.internal.subscriptions.SubscriptionHelper;
+import io.reactivex.plugins.RxJavaPlugins;
 
 public final class FlowableUnsubscribeOn<T> extends AbstractFlowableWithUpstream<T, T> {
     final Scheduler scheduler;
@@ -56,17 +57,25 @@ public void onSubscribe(Subscription s) {
 
         @Override
         public void onNext(T t) {
-            actual.onNext(t);
+            if (!get()) {
+                actual.onNext(t);
+            }
         }
 
         @Override
         public void onError(Throwable t) {
+            if (get()) {
+                RxJavaPlugins.onError(t);
+                return;
+            }
             actual.onError(t);
         }
 
         @Override
         public void onComplete() {
-            actual.onComplete();
+            if (!get()) {
+                actual.onComplete();
+            }
         }
 
         @Override

src/main/java/io/reactivex/internal/operators/observable/ObservableUnsubscribeOn.java

@@ -18,6 +18,7 @@
 import io.reactivex.*;
 import io.reactivex.disposables.Disposable;
 import io.reactivex.internal.disposables.DisposableHelper;
+import io.reactivex.plugins.RxJavaPlugins;
 
 public final class ObservableUnsubscribeOn<T> extends AbstractObservableWithUpstream<T, T> {
     final Scheduler scheduler;
@@ -55,17 +56,25 @@ public void onSubscribe(Disposable s) {
 
         @Override
         public void onNext(T t) {
-            actual.onNext(t);
+            if (!get()) {
+                actual.onNext(t);
+            }
         }
 
         @Override
         public void onError(Throwable t) {
+            if (get()) {
+                RxJavaPlugins.onError(t);
+                return;
+            }
             actual.onError(t);
         }
 
         @Override
         public void onComplete() {
-            actual.onComplete();
+            if (!get()) {
+                actual.onComplete();
+            }
         }
 
         @Override

src/test/java/io/reactivex/internal/operators/flowable/FlowableUnsubscribeOnTest.java

@@ -182,4 +182,17 @@ public Thread getThread() {
         }
 
     }
+
+    @Test
+    public void takeHalf() {
+        int elements = 1024;
+        Flowable.range(0, elements * 2).unsubscribeOn(Schedulers.single())
+        .take(elements)
+        .test()
+        .awaitDone(5, TimeUnit.SECONDS)
+        .assertValueCount(elements)
+        .assertComplete()
+        .assertNoErrors()
+        .assertSubscribed();
+    }
 }

src/test/java/io/reactivex/tck/AllTckTest.java

@@ -0,0 +1,41 @@
+/**
+ * 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.tck;
+
+import org.reactivestreams.Publisher;
+import org.testng.annotations.Test;
+
+import io.reactivex.Flowable;
+import io.reactivex.functions.Predicate;
+
+@Test
+public class AllTckTest extends BaseTck<Boolean> {
+
+    @Override
+    public Publisher<Boolean> createPublisher(final long elements) {
+        return FlowableTck.wrap(
+                Flowable.range(1, 1000).all(new Predicate<Integer>() {
+                    @Override
+                    public boolean test(Integer e) throws Exception {
+                        return e < 800;
+                    }
+                })
+            );
+    }
+
+    @Override
+    public long maxElementsFromPublisher() {
+        return 1;
+    }
+}

src/test/java/io/reactivex/tck/AmbArrayTckTest.java

@@ -0,0 +1,34 @@
+/**
+ * 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.tck;
+
+import org.reactivestreams.Publisher;
+import org.testng.annotations.Test;
+
+import io.reactivex.Flowable;
+
+@Test
+public class AmbArrayTckTest extends BaseTck<Long> {
+
+    @SuppressWarnings("unchecked")
+    @Override
+    public Publisher<Long> createPublisher(long elements) {
+        return FlowableTck.wrap(
+                Flowable.ambArray(
+                        Flowable.fromIterable(iterate(elements)),
+                        Flowable.<Long>never()
+                )
+            );
+    }
+}

src/test/java/io/reactivex/tck/AmbTckTest.java

@@ -0,0 +1,37 @@
+/**
+ * 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.tck;
+
+import java.util.Arrays;
+
+import org.reactivestreams.Publisher;
+import org.testng.annotations.Test;
+
+import io.reactivex.Flowable;
+
+@Test
+public class AmbTckTest extends BaseTck<Long> {
+
+    @SuppressWarnings("unchecked")
+    @Override
+    public Publisher<Long> createPublisher(long elements) {
+        return FlowableTck.wrap(
+            Flowable.amb(Arrays.asList(
+                    Flowable.fromIterable(iterate(elements)),
+                    Flowable.<Long>never()
+                )
+            )
+        );
+    }
+}

src/test/java/io/reactivex/tck/AnyTckTest.java

@@ -0,0 +1,41 @@
+/**
+ * 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.tck;
+
+import org.reactivestreams.Publisher;
+import org.testng.annotations.Test;
+
+import io.reactivex.Flowable;
+import io.reactivex.functions.Predicate;
+
+@Test
+public class AnyTckTest extends BaseTck<Boolean> {
+
+    @Override
+    public Publisher<Boolean> createPublisher(final long elements) {
+        return FlowableTck.wrap(
+                Flowable.range(1, 1000).any(new Predicate<Integer>() {
+                    @Override
+                    public boolean test(Integer e) throws Exception {
+                        return e == 500;
+                    }
+                })
+            );
+    }
+
+    @Override
+    public long maxElementsFromPublisher() {
+        return 1;
+    }
+}

src/test/java/io/reactivex/tck/BaseTck.java

@@ -25,19 +25,27 @@
 /**
  * Base abstract class for Flowable verifications, contains support for creating
  * Iterable range of values
+ * 
+ * @param <T> the element type
  */
 @Test
-public abstract class BaseTck extends PublisherVerification<Long> {
+public abstract class BaseTck<T> extends PublisherVerification<T> {
 
     public BaseTck() {
-        super(new TestEnvironment(300L));
+        super(new TestEnvironment(25L));
     }
 
     @Override
-    public Publisher<Long> createFailedPublisher() {
+    public Publisher<T> createFailedPublisher() {
         return Flowable.error(new TestException());
     }
 
+
+    @Override
+    public long maxElementsFromPublisher() {
+        return 1024;
+    }
+
     /**
      * Creates an Iterable with the specified number of elements or an infinite one if
      * elements > Integer.MAX_VALUE

src/test/java/io/reactivex/tck/BufferBoundaryTckTest.java

@@ -0,0 +1,39 @@
+/**
+ * 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.tck;
+
+import java.util.List;
+
+import org.reactivestreams.Publisher;
+import org.testng.annotations.Test;
+
+import io.reactivex.Flowable;
+
+@Test
+public class BufferBoundaryTckTest extends BaseTck<List<Long>> {
+
+    @Override
+    public Publisher<List<Long>> createPublisher(long elements) {
+        return FlowableTck.wrap(
+            Flowable.fromIterable(iterate(elements))
+            .buffer(Flowable.just(1).concatWith(Flowable.<Integer>never()))
+            .onBackpressureLatest()
+        );
+    }
+
+    @Override
+    public long maxElementsFromPublisher() {
+        return 1;
+    }
+}

src/test/java/io/reactivex/tck/BufferExactSizeTckTest.java

@@ -0,0 +1,33 @@
+/**
+ * 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.tck;
+
+import java.util.List;
+
+import org.reactivestreams.Publisher;
+import org.testng.annotations.Test;
+
+import io.reactivex.Flowable;
+
+@Test
+public class BufferExactSizeTckTest extends BaseTck<List<Long>> {
+
+    @Override
+    public Publisher<List<Long>> createPublisher(long elements) {
+        return FlowableTck.wrap(
+            Flowable.fromIterable(iterate(elements * 2))
+            .buffer(2)
+        );
+    }
+}

src/test/java/io/reactivex/tck/CacheTckTest.java

@@ -0,0 +1,30 @@
+/**
+ * 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.tck;
+
+import org.reactivestreams.Publisher;
+import org.testng.annotations.Test;
+
+import io.reactivex.Flowable;
+
+@Test
+public class CacheTckTest extends BaseTck<Long> {
+
+    @Override
+    public Publisher<Long> createPublisher(long elements) {
+        return FlowableTck.wrap(
+                Flowable.fromIterable(iterate(elements)).cache()
+            );
+    }
+}