Fixed ReplaySubject leak

Author: akarnokd

rxjava-core/src/main/java/rx/subjects/AsyncSubject.java

@@ -82,7 +82,7 @@ public void call(SubjectObserver<? super T> o) {
                         // to send onCompleted if the last value is an onNext
                         emitValueToObserver(lastNotification.get(), o);
                     }
-                });
+                }, null);
 
         return new AsyncSubject<T>(onSubscribe, subscriptionManager, lastNotification);
     }

rxjava-core/src/main/java/rx/subjects/BehaviorSubject.java

@@ -128,7 +128,7 @@ public void call(SubjectObserver<? super T> o) {
                          */
                         lastNotification.get().accept(o);
                     }
-                });
+                }, null);
 
         return new BehaviorSubject<T>(onSubscribe, subscriptionManager, lastNotification);
     }

rxjava-core/src/main/java/rx/subjects/PublishSubject.java

@@ -79,7 +79,7 @@ public void call(SubjectObserver<? super T> o) {
                          */
                         lastNotification.get().accept(o);
                     }
-                });
+                }, null);
 
         return new PublishSubject<T>(onSubscribe, subscriptionManager, lastNotification);
     }

rxjava-core/src/main/java/rx/subjects/ReplaySubject.java

@@ -84,8 +84,15 @@ public void call(SubjectObserver<? super T> o) {
 
                     @Override
                     public void call(SubjectObserver<? super T> o) {
+                        Integer idx = state.replayState.remove(o);
                         // we will finish replaying if there is anything left
-                        replayObserverFromIndex(state.history, state.replayState.get(o), o);
+                        replayObserverFromIndex(state.history, idx, o);
+                    }
+                }, 
+                new Action1<SubjectObserver<? super T>>() {
+                    @Override
+                    public void call(SubjectObserver<? super T> o) {
+                        state.replayState.remove(o);
                     }
                 });
 
@@ -229,5 +236,10 @@ public void complete(Notification<T> n) {
             terminalValue.set(n);
         }
     }
-
+    /**
+     * @return Returns the number of subscribers.
+     */
+    /* Support test.*/ int subscriberCount() {
+        return state.replayState.size();
+    }
 }

rxjava-core/src/main/java/rx/subjects/SubjectSubscriptionManager.java

@@ -39,13 +39,16 @@
      *            Always runs at the beginning of 'subscribe' regardless of terminal state.
      * @param onTerminated
      *            Only runs if Subject is in terminal state and the Observer ends up not being registered.
+     * @param onUnsubscribe called after the child subscription is removed from the state
      * @return
      */
-    public OnSubscribe<T> getOnSubscribeFunc(final Action1<SubjectObserver<? super T>> onSubscribe, final Action1<SubjectObserver<? super T>> onTerminated) {
+    public OnSubscribe<T> getOnSubscribeFunc(final Action1<SubjectObserver<? super T>> onSubscribe, 
+            final Action1<SubjectObserver<? super T>> onTerminated,
+            final Action1<SubjectObserver<? super T>> onUnsubscribe) {
         return new OnSubscribe<T>() {
             @Override
             public void call(Subscriber<? super T> actualOperator) {
-                SubjectObserver<T> observer = new SubjectObserver<T>(actualOperator);
+                final SubjectObserver<T> observer = new SubjectObserver<T>(actualOperator);
                 // invoke onSubscribe logic 
                 if (onSubscribe != null) {
                     onSubscribe.call(observer);
@@ -84,6 +87,9 @@ public void call() {
                                     // on unsubscribe remove it from the map of outbound observers to notify
                                     newState = current.removeObserver(subscription);
                                 } while (!state.compareAndSet(current, newState));
+                                if (onUnsubscribe != null) {
+                                    onUnsubscribe.call(observer);
+                                }
                             }
                         }));
 

rxjava-core/src/test/java/rx/subjects/ReplaySubjectTest.java

@@ -344,5 +344,16 @@ public void onNext(String v) {
         assertEquals("three", lastValueForObserver2.get());
 
     }
+    @Test
+    public void testSubscriptionLeak() {
+        ReplaySubject<Object> replaySubject = ReplaySubject.create();
+        
+        Subscription s = replaySubject.subscribe();
+
+        assertEquals(1, replaySubject.subscriberCount());
 
+        s.unsubscribe();
+        
+        assertEquals(0, replaySubject.subscriberCount());
+    }
 }