Merge pull request #894 from benjchristensen/synchronize-isTerminated

Synchronized Operator Check for isTerminated

Author: benjchristensen

rxjava-core/src/main/java/rx/observers/SynchronizedObserver.java

@@ -41,7 +41,8 @@ public final class SynchronizedObserver<T> implements Observer<T> {
      */
 
     private final Observer<? super T> observer;
-    private volatile Object lock;
+    private final Object lock;
+    private boolean isTerminated = false;
 
     public SynchronizedObserver(Observer<? super T> subscriber) {
         this.observer = subscriber;
@@ -55,19 +56,27 @@ public SynchronizedObserver(Observer<? super T> subscriber, Object lock) {
 
     public void onNext(T arg) {
         synchronized (lock) {
-            observer.onNext(arg);
+            if (!isTerminated) {
+                observer.onNext(arg);
+            }
         }
     }
 
     public void onError(Throwable e) {
         synchronized (lock) {
-            observer.onError(e);
+            if (!isTerminated) {
+                isTerminated = true;
+                observer.onError(e);
+            }
         }
     }
 
     public void onCompleted() {
         synchronized (lock) {
-            observer.onCompleted();
+            if (!isTerminated) {
+                isTerminated = true;
+                observer.onCompleted();
+            }
         }
     }
 }