Merge pull request #3333 from akarnokd/SubscribersTests2x

2.x: subscribers/observers tests

Author: akarnokd

src/main/java/io/reactivex/exceptions/CompositeException.java

@@ -0,0 +1,44 @@
+/**
+ * Copyright 2015 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.exceptions;
+
+import java.util.*;
+
+public final class CompositeException extends RuntimeException {
+    /** */
+    private static final long serialVersionUID = 2004635183691362481L;
+
+    public CompositeException() {
+        super();
+    }
+
+    public CompositeException(String message) {
+        super(message);
+    }
+
+    public List<Throwable> getExceptions() {
+        Throwable cause = getCause();
+        Throwable[] suppressed = getSuppressed();
+        List<Throwable> list = new ArrayList<>(cause != null 
+                ? 1 + suppressed.length : suppressed.length);
+        if (cause != null) {
+            list.add(cause);
+        }
+        for (Throwable t : suppressed) {
+            list.add(t);
+        }
+        
+        return list;
+    }
+}

src/main/java/io/reactivex/exceptions/OnCompleteFailedException.java

@@ -0,0 +1,24 @@
+/**
+ * Copyright 2015 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.exceptions;
+
+public final class OnCompleteFailedException extends RuntimeException {
+    /** */
+    private static final long serialVersionUID = -6179993283427447098L;
+
+    public OnCompleteFailedException(Throwable cause) {
+        super(cause);
+    }
+    
+}

src/main/java/io/reactivex/exceptions/OnErrorFailedException.java

@@ -0,0 +1,23 @@
+/**
+ * Copyright 2015 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.exceptions;
+
+public final class OnErrorFailedException extends RuntimeException {
+    /** */
+    private static final long serialVersionUID = 2656125445290831911L;
+
+    public OnErrorFailedException(Throwable cause) {
+        super(cause);
+    }
+}

src/main/java/io/reactivex/exceptions/OnErrorNotImplementedException.java

@@ -0,0 +1,36 @@
+/**
+ * Copyright 2015 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.exceptions;
+
+public final class OnErrorNotImplementedException extends RuntimeException {
+    /** */
+    private static final long serialVersionUID = -3698670655303683299L;
+
+    public OnErrorNotImplementedException() {
+        super();
+    }
+
+    public OnErrorNotImplementedException(String message, Throwable cause) {
+        super(message, cause);
+    }
+
+    public OnErrorNotImplementedException(String message) {
+        super(message);
+    }
+
+    public OnErrorNotImplementedException(Throwable cause) {
+        super(cause);
+    }
+    
+}

src/main/java/io/reactivex/exceptions/UnsubscribeFailedException.java

@@ -0,0 +1,24 @@
+/**
+ * Copyright 2015 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.exceptions;
+
+public final class UnsubscribeFailedException extends RuntimeException {
+    /** */
+    private static final long serialVersionUID = 8947024194181365640L;
+
+    public UnsubscribeFailedException(Throwable cause) {
+        super(cause);
+    }
+    
+}

src/main/java/io/reactivex/internal/operators/OperatorDoOnEach.java

@@ -51,6 +51,8 @@ static final class DoOnEachSubscriber<T> implements Subscriber<T> {
 
         Subscription s;
 
+        boolean done;
+        
         public DoOnEachSubscriber(
                 Subscriber<? super T> actual,
                 Consumer<? super T> onNext, 
@@ -75,6 +77,9 @@ public void onSubscribe(Subscription s) {
 
         @Override
         public void onNext(T t) {
+            if (done) {
+                return;
+            }
             try {
                 onNext.accept(t);
             } catch (Throwable e) {
@@ -88,6 +93,11 @@ public void onNext(T t) {
 
         @Override
         public void onError(Throwable t) {
+            if (done) {
+                RxJavaPlugins.onError(t);
+                return;
+            }
+            done = true;
             try {
                 onError.accept(t);
             } catch (Throwable e) {
@@ -104,6 +114,10 @@ public void onError(Throwable t) {
 
         @Override
         public void onComplete() {
+            if (done) {
+                return;
+            }
+            done = true;
             try {
                 onComplete.run();
             } catch (Throwable e) {

src/main/java/io/reactivex/internal/subscribers/CancelledSubscriber.java

@@ -18,7 +18,7 @@
 import io.reactivex.plugins.RxJavaPlugins;
 
 /**
- * A subscriber cancels the subscription sent to it 
+ * A subscriber that cancels the subscription sent to it 
  * and ignores all events (onError is forwarded to RxJavaPlugins though).
  */
 public enum CancelledSubscriber implements Subscriber<Object> {