diff --git a/rxjava-core/src/main/java/rx/joins/JoinObserver1.java b/rxjava-core/src/main/java/rx/joins/JoinObserver1.java
index ede55f0584..b191641352 100644
--- a/rxjava-core/src/main/java/rx/joins/JoinObserver1.java
+++ b/rxjava-core/src/main/java/rx/joins/JoinObserver1.java
@@ -23,28 +23,31 @@
 
 import rx.Notification;
 import rx.Observable;
+import rx.Observer;
 import rx.operators.SafeObservableSubscription;
+import rx.operators.SafeObserver;
 import rx.util.functions.Action1;
 
 /**
  * Default implementation of a join observer.
  */
-public final class JoinObserver1<T> extends ObserverBase<Notification<T>> implements JoinObserver {
+public final class JoinObserver1<T> implements Observer<Notification<T>>, JoinObserver {
     private Object gate;
     private final Observable<T> source;
     private final Action1<Throwable> onError;
     private final List<ActivePlan0> activePlans;
     private final Queue<Notification<T>> queue;
-    private final SafeObservableSubscription subscription;
+    private final SafeObservableSubscription subscription = new SafeObservableSubscription();
     private volatile boolean done;
     private final AtomicBoolean subscribed = new AtomicBoolean(false);
+    private final SafeObserver<Notification<T>> safeObserver;
     
     public JoinObserver1(Observable<T> source, Action1<Throwable> onError) {
         this.source = source;
         this.onError = onError;
         queue = new LinkedList<Notification<T>>();
-        subscription = new SafeObservableSubscription();
         activePlans = new ArrayList<ActivePlan0>();
+        safeObserver = new SafeObserver<Notification<T>>(subscription, new InnerObserver());
     }
     public Queue<Notification<T>> queue() {
         return queue;
@@ -67,35 +70,52 @@ public void dequeue() {
         queue.remove();
     }
 
-    @Override
-    protected void onNextCore(Notification<T> args) {
-        synchronized (gate) {
-            if (!done) {
-                if (args.isOnError()) {
-                    onError.call(args.getThrowable());
-                    return;
-                }
-                queue.add(args);
-                
-                // remark: activePlans might change while iterating
-                for (ActivePlan0 a : new ArrayList<ActivePlan0>(activePlans)) {
-                    a.match();
+    private final class InnerObserver implements Observer<Notification<T>> {
+
+        @Override
+        public void onNext(Notification<T> args) {
+            synchronized (gate) {
+                if (!done) {
+                    if (args.isOnError()) {
+                        onError.call(args.getThrowable());
+                        return;
+                    }
+                    queue.add(args);
+
+                    // remark: activePlans might change while iterating
+                    for (ActivePlan0 a : new ArrayList<ActivePlan0>(activePlans)) {
+                        a.match();
+                    }
                 }
             }
         }
+
+        @Override
+        public void onError(Throwable e) {
+            // not expected
+        }
+
+        @Override
+        public void onCompleted() {
+            // not expected or ignored
+        }
+    }
+    
+    @Override
+    public void onNext(Notification<T> args) {
+        safeObserver.onNext(args);
     }
 
     @Override
-    protected void onErrorCore(Throwable e) {
-        // not expected
+    public void onError(Throwable e) {
+        safeObserver.onError(e);
     }
 
     @Override
-    protected void onCompletedCore() {
-        // not expected or ignored
+    public void onCompleted() {
+        safeObserver.onCompleted();
     }
     
-    
     void removeActivePlan(ActivePlan0 activePlan) {
         activePlans.remove(activePlan);
         if (activePlans.isEmpty()) {
diff --git a/rxjava-core/src/main/java/rx/joins/ObserverBase.java b/rxjava-core/src/main/java/rx/joins/ObserverBase.java
deleted file mode 100644
index f1144a8ad2..0000000000
--- a/rxjava-core/src/main/java/rx/joins/ObserverBase.java
+++ /dev/null
@@ -1,72 +0,0 @@
-/**
- * Copyright 2013 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 rx.joins;
-
-import java.util.concurrent.atomic.AtomicBoolean;
-import rx.Observer;
-
-/**
- * Implements an observer that ensures proper event delivery
- * semantics to its abstract onXxxxCore methods.
- */
-public abstract class ObserverBase<T> implements Observer<T> {
-    private final AtomicBoolean completed = new AtomicBoolean();
-
-    @Override
-    public void onNext(T args) {
-        if (!completed.get()) {
-            onNextCore(args);
-        }
-    }
-
-    @Override
-    public void onError(Throwable e) {
-        if (completed.compareAndSet(false, true)) {
-            onErrorCore(e);
-        }
-    }
-
-    @Override
-    public void onCompleted() {
-        if (completed.compareAndSet(false, true)) {
-            onCompletedCore();
-        }
-    }
-    /**
-     * Implement this method to react to the receival of a new element in the sequence.
-     */
-    protected abstract void onNextCore(T args);
-    /**
-     * Implement this method to react to the occurrence of an exception.
-     */
-    protected abstract void onErrorCore(Throwable e);
-    /**
-     * Implement this method to react to the end of the sequence.
-     */
-    protected abstract void onCompletedCore();
-    /**
-     * Try to trigger the error state.
-     * @param t 
-     * @return false if already completed
-     */
-    protected boolean fail(Throwable t) {
-        if (completed.compareAndSet(false, true)) {
-            onErrorCore(t);
-            return true;
-        }
-        return false;
-    }
-}