diff --git a/src/main/java/rx/subjects/ReplaySubject.java b/src/main/java/rx/subjects/ReplaySubject.java
index 4c83db51341..a137dd6c4d7 100644
--- a/src/main/java/rx/subjects/ReplaySubject.java
+++ b/src/main/java/rx/subjects/ReplaySubject.java
@@ -102,6 +102,42 @@ public void call(SubjectObserver<T> o) {
                 o.index(lastIndex);
             }
         };
+        ssm.onAdded = new Action1<SubjectObserver<T>>() {
+            @Override
+            public void call(SubjectObserver<T> o) {
+                synchronized (o) {
+                    if (!o.first || o.emitting) {
+                        return;
+                    }
+                    o.first = false;
+                    o.emitting = true;
+                }
+                boolean skipFinal = false;
+                try {
+                    for (;;) {
+                        int idx = o.index();
+                        int sidx = state.index;
+                        if (idx != sidx) {
+                            Integer j = state.replayObserverFromIndex(idx, o);
+                            o.index(j);
+                        }
+                        synchronized (o) {
+                            if (sidx == state.index) {
+                                o.emitting = false;
+                                skipFinal = true;
+                                break;
+                            }
+                        }
+                    }
+                } finally {
+                    if (!skipFinal) {
+                        synchronized (o) {
+                            o.emitting = false;
+                        }
+                    }
+                }
+            }
+        };
         ssm.onTerminated = new Action1<SubjectObserver<T>>() {
             @Override
             public void call(SubjectObserver<T> o) {
@@ -264,6 +300,42 @@ static final <T> ReplaySubject<T> createWithState(final BoundedState<T> state,
             Action1<SubjectObserver<T>> onStart) {
         SubjectSubscriptionManager<T> ssm = new SubjectSubscriptionManager<T>();
         ssm.onStart = onStart;
+        ssm.onAdded = new Action1<SubjectObserver<T>>() {
+            @Override
+            public void call(SubjectObserver<T> o) {
+                synchronized (o) {
+                    if (!o.first || o.emitting) {
+                        return;
+                    }
+                    o.first = false;
+                    o.emitting = true;
+                }
+                boolean skipFinal = false;
+                try {
+                    for (;;) {
+                        NodeList.Node<Object> idx = o.index();
+                        NodeList.Node<Object> sidx = state.tail();
+                        if (idx != sidx) {
+                            NodeList.Node<Object> j = state.replayObserverFromIndex(idx, o);
+                            o.index(j);
+                        }
+                        synchronized (o) {
+                            if (sidx == state.tail()) {
+                                o.emitting = false;
+                                skipFinal = true;
+                                break;
+                            }
+                        }
+                    }
+                } finally {
+                    if (!skipFinal) {
+                        synchronized (o) {
+                            o.emitting = false;
+                        }
+                    }
+                }
+            }
+        };
         ssm.onTerminated = new Action1<SubjectObserver<T>>() {
 
             @Override
@@ -355,9 +427,10 @@ public boolean hasObservers() {
 
     private boolean caughtUp(SubjectObserver<? super T> o) {
         if (!o.caughtUp) {
-            o.caughtUp = true;
-            state.replayObserver(o);
-            o.index(null); // once caught up, no need for the index anymore
+            if (state.replayObserver(o)) {
+                o.caughtUp = true;
+                o.index(null); // once caught up, no need for the index anymore
+            }
             return false;
         } else {
             // it was caught up so proceed the "raw route"
@@ -423,11 +496,20 @@ public boolean terminated() {
         }
 
         @Override
-        public void replayObserver(SubjectObserver<? super T> observer) {
+        public boolean replayObserver(SubjectObserver<? super T> observer) {
+            
+            synchronized (observer) {
+                observer.first = false;
+                if (observer.emitting) {
+                    return false;
+                }
+            }
+            
             Integer lastEmittedLink = observer.index();
             if (lastEmittedLink != null) {
                 int l = replayObserverFromIndex(lastEmittedLink, observer);
                 observer.index(l);
+                return true;
             } else {
                 throw new IllegalStateException("failed to find lastEmittedLink for: " + observer);
             }
@@ -525,10 +607,18 @@ public Node<Object> tail() {
             return tail;
         }
         @Override
-        public void replayObserver(SubjectObserver<? super T> observer) {
+        public boolean replayObserver(SubjectObserver<? super T> observer) {
+            synchronized (observer) {
+                observer.first = false;
+                if (observer.emitting) {
+                    return false;
+                }
+            }
+            
             NodeList.Node<Object> lastEmittedLink = observer.index();
             NodeList.Node<Object> l = replayObserverFromIndex(lastEmittedLink, observer);
             observer.index(l);
+            return true;
         }
 
         @Override
@@ -571,8 +661,9 @@ interface ReplayState<T, I> {
         /**
          * Replay contents to the given observer.
          * @param observer the receiver of events
+         * @return true if the subject has caught up
          */
-        void replayObserver(SubjectObserver<? super T> observer);
+        boolean replayObserver(SubjectObserver<? super T> observer);
         /**
          * Replay the buffered values from an index position and return a new index
          * @param idx the current index position
diff --git a/src/test/java/rx/subjects/BehaviorSubjectTest.java b/src/test/java/rx/subjects/BehaviorSubjectTest.java
index aa5f7164819..76ae5bd7234 100644
--- a/src/test/java/rx/subjects/BehaviorSubjectTest.java
+++ b/src/test/java/rx/subjects/BehaviorSubjectTest.java
@@ -25,17 +25,19 @@
 import static org.mockito.Mockito.times;
 import static org.mockito.Mockito.verify;
 
-import org.junit.Test;
+import java.util.concurrent.*;
+import java.util.concurrent.atomic.AtomicReference;
+
+import org.junit.*;
 import org.mockito.InOrder;
 import org.mockito.Mockito;
 
-import rx.Observable;
-import rx.Observer;
-import rx.Subscription;
+import rx.*;
 import rx.exceptions.CompositeException;
 import rx.exceptions.OnErrorNotImplementedException;
-import rx.functions.Func1;
+import rx.functions.*;
 import rx.observers.TestSubscriber;
+import rx.schedulers.Schedulers;
 
 public class BehaviorSubjectTest {
 
@@ -417,4 +419,67 @@ public void testOnErrorThrowsDoesntPreventDelivery2() {
         // even though the onError above throws we should still receive it on the other subscriber 
         assertEquals(1, ts.getOnErrorEvents().size());
     }
+    @Test
+    public void testEmissionSubscriptionRace() throws Exception {
+        Scheduler s = Schedulers.io();
+        Scheduler.Worker worker = Schedulers.io().createWorker();
+        for (int i = 0; i < 50000; i++) {
+            if (i % 1000 == 0) {
+                System.out.println(i);
+            }
+            final BehaviorSubject<Object> rs = BehaviorSubject.create();
+            
+            final CountDownLatch finish = new CountDownLatch(1); 
+            final CountDownLatch start = new CountDownLatch(1); 
+            
+            worker.schedule(new Action0() {
+                @Override
+                public void call() {
+                    try {
+                        start.await();
+                    } catch (Exception e1) {
+                        e1.printStackTrace();
+                    }
+                    rs.onNext(1);
+                }
+            });
+            
+            final AtomicReference<Object> o = new AtomicReference<Object>();
+            
+            rs.subscribeOn(s).observeOn(Schedulers.io())
+            .subscribe(new Observer<Object>() {
+
+                @Override
+                public void onCompleted() {
+                    o.set(-1);
+                    finish.countDown();
+                }
+
+                @Override
+                public void onError(Throwable e) {
+                    o.set(e);
+                    finish.countDown();
+                }
+
+                @Override
+                public void onNext(Object t) {
+                    o.set(t);
+                    finish.countDown();
+                }
+                
+            });
+            start.countDown();
+            
+            if (!finish.await(5, TimeUnit.SECONDS)) {
+                System.out.println(o.get());
+                System.out.println(rs.hasObservers());
+                rs.onCompleted();
+                Assert.fail("Timeout @ " + i);
+                break;
+            } else {
+                Assert.assertEquals(1, o.get());
+                rs.onCompleted();
+            }
+        }
+    }
 }
diff --git a/src/test/java/rx/subjects/ReplaySubjectBoundedConcurrencyTest.java b/src/test/java/rx/subjects/ReplaySubjectBoundedConcurrencyTest.java
index d086224b9ce..bf4c2bbc1e0 100644
--- a/src/test/java/rx/subjects/ReplaySubjectBoundedConcurrencyTest.java
+++ b/src/test/java/rx/subjects/ReplaySubjectBoundedConcurrencyTest.java
@@ -25,12 +25,12 @@
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicReference;
 
-import org.junit.Test;
+import org.junit.*;
 
-import rx.Observable;
+import rx.*;
 import rx.Observable.OnSubscribe;
 import rx.Subscriber;
-import rx.functions.Action1;
+import rx.functions.*;
 import rx.observers.TestSubscriber;
 import rx.schedulers.Schedulers;
 
@@ -337,4 +337,67 @@ public void run() {
             }
         }
     }
+    @Test
+    public void testReplaySubjectEmissionSubscriptionRace() throws Exception {
+        Scheduler s = Schedulers.io();
+        Scheduler.Worker worker = Schedulers.io().createWorker();
+        for (int i = 0; i < 50000; i++) {
+            if (i % 1000 == 0) {
+                System.out.println(i);
+            }
+            final ReplaySubject<Object> rs = ReplaySubject.createWithSize(2);
+            
+            final CountDownLatch finish = new CountDownLatch(1); 
+            final CountDownLatch start = new CountDownLatch(1); 
+            
+            worker.schedule(new Action0() {
+                @Override
+                public void call() {
+                    try {
+                        start.await();
+                    } catch (Exception e1) {
+                        e1.printStackTrace();
+                    }
+                    rs.onNext(1);
+                }
+            });
+            
+            final AtomicReference<Object> o = new AtomicReference<Object>();
+            
+            rs.subscribeOn(s).observeOn(Schedulers.io())
+            .subscribe(new Observer<Object>() {
+
+                @Override
+                public void onCompleted() {
+                    o.set(-1);
+                    finish.countDown();
+                }
+
+                @Override
+                public void onError(Throwable e) {
+                    o.set(e);
+                    finish.countDown();
+                }
+
+                @Override
+                public void onNext(Object t) {
+                    o.set(t);
+                    finish.countDown();
+                }
+                
+            });
+            start.countDown();
+            
+            if (!finish.await(5, TimeUnit.SECONDS)) {
+                System.out.println(o.get());
+                System.out.println(rs.hasObservers());
+                rs.onCompleted();
+                Assert.fail("Timeout @ " + i);
+                break;
+            } else {
+                Assert.assertEquals(1, o.get());
+                rs.onCompleted();
+            }
+        }
+    }
 }
\ No newline at end of file
diff --git a/src/test/java/rx/subjects/ReplaySubjectConcurrencyTest.java b/src/test/java/rx/subjects/ReplaySubjectConcurrencyTest.java
index 4c7c9477451..304533e0129 100644
--- a/src/test/java/rx/subjects/ReplaySubjectConcurrencyTest.java
+++ b/src/test/java/rx/subjects/ReplaySubjectConcurrencyTest.java
@@ -17,20 +17,17 @@
 
 import static org.junit.Assert.assertEquals;
 
-import java.util.ArrayList;
-import java.util.Arrays;
-import java.util.Collections;
-import java.util.List;
-import java.util.concurrent.CountDownLatch;
-import java.util.concurrent.TimeUnit;
+import java.util.*;
+import java.util.concurrent.*;
 import java.util.concurrent.atomic.AtomicReference;
 
-import org.junit.Test;
+import org.junit.*;
 
-import rx.Observable;
+import rx.*;
 import rx.Observable.OnSubscribe;
-import rx.Subscriber;
-import rx.functions.Action1;
+import rx.Observable;
+import rx.Observer;
+import rx.functions.*;
 import rx.observers.TestSubscriber;
 import rx.schedulers.Schedulers;
 
@@ -337,4 +334,67 @@ public void run() {
             }
         }
     }
+    @Test
+    public void testReplaySubjectEmissionSubscriptionRace() throws Exception {
+        Scheduler s = Schedulers.io();
+        Scheduler.Worker worker = Schedulers.io().createWorker();
+        for (int i = 0; i < 50000; i++) {
+            if (i % 1000 == 0) {
+                System.out.println(i);
+            }
+            final ReplaySubject<Object> rs = ReplaySubject.create();
+            
+            final CountDownLatch finish = new CountDownLatch(1); 
+            final CountDownLatch start = new CountDownLatch(1); 
+            
+            worker.schedule(new Action0() {
+                @Override
+                public void call() {
+                    try {
+                        start.await();
+                    } catch (Exception e1) {
+                        e1.printStackTrace();
+                    }
+                    rs.onNext(1);
+                }
+            });
+            
+            final AtomicReference<Object> o = new AtomicReference<Object>();
+            
+            rs.subscribeOn(s).observeOn(Schedulers.io())
+            .subscribe(new Observer<Object>() {
+
+                @Override
+                public void onCompleted() {
+                    o.set(-1);
+                    finish.countDown();
+                }
+
+                @Override
+                public void onError(Throwable e) {
+                    o.set(e);
+                    finish.countDown();
+                }
+
+                @Override
+                public void onNext(Object t) {
+                    o.set(t);
+                    finish.countDown();
+                }
+                
+            });
+            start.countDown();
+            
+            if (!finish.await(5, TimeUnit.SECONDS)) {
+                System.out.println(o.get());
+                System.out.println(rs.hasObservers());
+                rs.onCompleted();
+                Assert.fail("Timeout @ " + i);
+                break;
+            } else {
+                Assert.assertEquals(1, o.get());
+                rs.onCompleted();
+            }
+        }
+    }
 }