Fix Subject.toObservable

benjchristensen · benjchristensen · commit f5eb8f77d66b · 2014-01-25T08:33:20.000-08:00
- now correctly creates only 1 Observable instance for the life of the Subject - this fixes the OperationJoinsTest - thanks @akarnokd for pointing out my mistake! - all rxjava-core unit tests are now passing
diff --git a/rxjava-core/src/main/java/rx/subjects/AsyncSubject.java b/rxjava-core/src/main/java/rx/subjects/AsyncSubject.java
@@ -96,19 +96,19 @@ protected static <T> void emitValueToObserver(Notification<T> n, Observer<? supe
         }
     }
 
-    private final OnSubscribe<T> onSubscribe;
     private final SubjectSubscriptionManager<T> subscriptionManager;
     final AtomicReference<Notification<T>> lastNotification;
+    private final Observable<T> observable;
 
     protected AsyncSubject(OnSubscribe<T> onSubscribe, SubjectSubscriptionManager<T> subscriptionManager, AtomicReference<Notification<T>> lastNotification) {
-        this.onSubscribe = onSubscribe;
         this.subscriptionManager = subscriptionManager;
         this.lastNotification = lastNotification;
+        this.observable = Observable.create(onSubscribe);
     }
 
     @Override
     public Observable<T> toObservable() {
-        return Observable.create(onSubscribe);
+        return observable;
     }
 
     @Override
@@ -130,7 +130,7 @@ public void onError(final Throwable e) {
 
             @Override
             public void call(Collection<SubjectObserver<? super T>> observers) {
-                lastNotification.set(new Notification<T>(e));
+                lastNotification.set(Notification.<T>createOnError(e));
                 for (Observer<? super T> o : observers) {
                     emitValueToObserver(lastNotification.get(), o);
                 }
@@ -141,7 +141,7 @@ public void call(Collection<SubjectObserver<? super T>> observers) {
 
     @Override
     public void onNext(T v) {
-        lastNotification.set(new Notification<T>(v));
+        lastNotification.set(Notification.<T>createOnNext(v));
     }
 
 }
diff --git a/rxjava-core/src/main/java/rx/subjects/BehaviorSubject.java b/rxjava-core/src/main/java/rx/subjects/BehaviorSubject.java
@@ -119,19 +119,19 @@ public void call(SubjectObserver<? super T> o) {
         return new BehaviorSubject<T>(onSubscribe, subscriptionManager, lastNotification);
     }
 
-    private final OnSubscribe<T> onSubscribe;
     private final SubjectSubscriptionManager<T> subscriptionManager;
     final AtomicReference<Notification<T>> lastNotification;
+    private final Observable<T> observable;
 
     protected BehaviorSubject(OnSubscribe<T> onSubscribe, SubjectSubscriptionManager<T> subscriptionManager, AtomicReference<Notification<T>> lastNotification) {
-        this.onSubscribe = onSubscribe;
         this.subscriptionManager = subscriptionManager;
         this.lastNotification = lastNotification;
+        this.observable = Observable.create(onSubscribe);
     }
 
     @Override
     public Observable<T> toObservable() {
-        return Observable.create(onSubscribe);
+        return observable;
     }
 
     @Override
@@ -140,7 +140,7 @@ public void onCompleted() {
 
             @Override
             public void call(Collection<SubjectObserver<? super T>> observers) {
-                lastNotification.set(new Notification<T>());
+                lastNotification.set(Notification.<T>createOnCompleted());
                 for (Observer<? super T> o : observers) {
                     o.onCompleted();
                 }
@@ -154,7 +154,7 @@ public void onError(final Throwable e) {
 
             @Override
             public void call(Collection<SubjectObserver<? super T>> observers) {
-                lastNotification.set(new Notification<T>(e));
+                lastNotification.set(Notification.<T>createOnError(e));
                 for (Observer<? super T> o : observers) {
                     o.onError(e);
                 }
@@ -168,7 +168,7 @@ public void onNext(T v) {
         // do not overwrite a terminal notification
         // so new subscribers can get them
         if (lastNotification.get().isOnNext()) {
-            lastNotification.set(new Notification<T>(v));
+            lastNotification.set(Notification.<T>createOnNext(v));
             for (Observer<? super T> o : subscriptionManager.rawSnapshot()) {
                 o.onNext(v);
             }
diff --git a/rxjava-core/src/main/java/rx/subjects/PublishSubject.java b/rxjava-core/src/main/java/rx/subjects/PublishSubject.java
@@ -86,19 +86,19 @@ public void call(SubjectObserver<? super T> o) {
         return new PublishSubject<T>(onSubscribe, subscriptionManager, lastNotification);
     }
 
-    private final OnSubscribe<T> onSubscribe;
     private final SubjectSubscriptionManager<T> subscriptionManager;
     final AtomicReference<Notification<T>> lastNotification;
+    private final Observable<T> observable;
 
     protected PublishSubject(OnSubscribe<T> onSubscribe, SubjectSubscriptionManager<T> subscriptionManager, AtomicReference<Notification<T>> lastNotification) {
-        this.onSubscribe = onSubscribe;
         this.subscriptionManager = subscriptionManager;
         this.lastNotification = lastNotification;
+        this.observable = Observable.create(onSubscribe);
     }
 
     @Override
     public Observable<T> toObservable() {
-        return Observable.create(onSubscribe);
+        return observable;
     }
 
     @Override
@@ -107,7 +107,7 @@ public void onCompleted() {
 
             @Override
             public void call(Collection<SubjectObserver<? super T>> observers) {
-                lastNotification.set(new Notification<T>());
+                lastNotification.set(Notification.<T>createOnCompleted());
                 for (Observer<? super T> o : observers) {
                     o.onCompleted();
                 }
@@ -121,7 +121,7 @@ public void onError(final Throwable e) {
 
             @Override
             public void call(Collection<SubjectObserver<? super T>> observers) {
-                lastNotification.set(new Notification<T>(e));
+                lastNotification.set(Notification.<T>createOnError(e));
                 for (Observer<? super T> o : observers) {
                     o.onError(e);
                 }
diff --git a/rxjava-core/src/main/java/rx/subjects/ReplaySubject.java b/rxjava-core/src/main/java/rx/subjects/ReplaySubject.java
@@ -108,17 +108,17 @@ public ReplayState(int initialCapacity) {
 
     private final SubjectSubscriptionManager<T> subscriptionManager;
     private final ReplayState<T> state;
-    private final OnSubscribe<T> onSubscribe;
+    private final Observable<T> observable;
 
     protected ReplaySubject(OnSubscribe<T> onSubscribe, SubjectSubscriptionManager<T> subscriptionManager, ReplayState<T> state) {
-        this.onSubscribe = onSubscribe;
         this.subscriptionManager = subscriptionManager;
         this.state = state;
+        this.observable = Observable.create(onSubscribe);
     }
 
     @Override
     public Observable<T> toObservable() {
-        return Observable.create(onSubscribe);
+        return observable;
     }
 
     @Override
@@ -127,7 +127,7 @@ public void onCompleted() {
 
             @Override
             public void call(Collection<SubjectObserver<? super T>> observers) {
-                state.history.complete(new Notification<T>());
+                state.history.complete(Notification.<T>createOnCompleted());
                 for (SubjectObserver<? super T> o : observers) {
                     if (caughtUp(o)) {
                         o.onCompleted();
@@ -143,7 +143,7 @@ public void onError(final Throwable e) {
 
             @Override
             public void call(Collection<SubjectObserver<? super T>> observers) {
-                state.history.complete(new Notification<T>(e));
+                state.history.complete(Notification.<T>createOnError(e));
                 for (SubjectObserver<? super T> o : observers) {
                     if (caughtUp(o)) {
                         o.onError(e);

Original file line number	Diff line number	Diff line change
`@@ -96,19 +96,19 @@ protected static <T> void emitValueToObserver(Notification<T> n, Observer<? supe`
`96`	`96`	`}`
`97`	`97`	`}`
`98`	`98`
`99`		`- private final OnSubscribe<T> onSubscribe;`
`100`	`99`	`private final SubjectSubscriptionManager<T> subscriptionManager;`
`101`	`100`	`final AtomicReference<Notification<T>> lastNotification;`
	`101`	`+ private final Observable<T> observable;`
`102`	`102`
`103`	`103`	`protected AsyncSubject(OnSubscribe<T> onSubscribe, SubjectSubscriptionManager<T> subscriptionManager, AtomicReference<Notification<T>> lastNotification) {`
`104`		`- this.onSubscribe = onSubscribe;`
`105`	`104`	`this.subscriptionManager = subscriptionManager;`
`106`	`105`	`this.lastNotification = lastNotification;`
	`106`	`+ this.observable = Observable.create(onSubscribe);`
`107`	`107`	`}`
`108`	`108`
`109`	`109`	`@Override`
`110`	`110`	`public Observable<T> toObservable() {`
`111`		`- return Observable.create(onSubscribe);`
	`111`	`+ return observable;`
`112`	`112`	`}`
`113`	`113`
`114`	`114`	`@Override`
`@@ -130,7 +130,7 @@ public void onError(final Throwable e) {`
`130`	`130`
`131`	`131`	`@Override`
`132`	`132`	`public void call(Collection<SubjectObserver<? super T>> observers) {`
`133`		`- lastNotification.set(new Notification<T>(e));`
	`133`	`+ lastNotification.set(Notification.<T>createOnError(e));`
`134`	`134`	`for (Observer<? super T> o : observers) {`
`135`	`135`	`emitValueToObserver(lastNotification.get(), o);`
`136`	`136`	`}`
`@@ -141,7 +141,7 @@ public void call(Collection<SubjectObserver<? super T>> observers) {`
`141`	`141`
`142`	`142`	`@Override`
`143`	`143`	`public void onNext(T v) {`
`144`		`- lastNotification.set(new Notification<T>(v));`
	`144`	`+ lastNotification.set(Notification.<T>createOnNext(v));`
`145`	`145`	`}`
`146`	`146`
`147`	`147`	`}`