Rename Subject to PublishSubject

Author: johngmyers

rxjava-core/src/main/java/rx/operators/OperationTake.java

@@ -26,10 +26,11 @@
 import rx.Observable;
 import rx.Observer;
 import rx.Subscription;
+import rx.subjects.PublishSubject;
 import rx.util.AtomicObservableSubscription;
 import rx.util.functions.Func1;
 import rx.util.functions.Func2;
-import rx.subjects.Subject;
+
 /**
  * Returns a specified number of contiguous values from the start of an observable sequence.
  */
@@ -181,11 +182,13 @@ public Boolean call(Integer input) {
 
         @Test
         public void testTakeWhileOnSubject1() {
-            Subject<Integer> s = Subject.create();
+            PublishSubject<Integer> s = PublishSubject.create();
             Observable<Integer> w = (Observable<Integer>)s;
-            Observable<Integer> take = Observable.create(takeWhile(w, new Func1<Integer, Boolean>() {
+            Observable<Integer> take = Observable.create(takeWhile(w, new Func1<Integer, Boolean>()
+            {
                 @Override
-                public Boolean call(Integer input) {
+                public Boolean call(Integer input)
+                {
                     return input < 3;
                 }
             }));
@@ -213,9 +216,11 @@ public Boolean call(Integer input) {
         @Test
         public void testTakeWhile2() {
             Observable<String> w = Observable.toObservable("one", "two", "three");
-            Observable<String> take = Observable.create(takeWhileWithIndex(w, new Func2<String, Integer, Boolean>() {
+            Observable<String> take = Observable.create(takeWhileWithIndex(w, new Func2<String, Integer, Boolean>()
+            {
                 @Override
-                public Boolean call(String input, Integer index) {
+                public Boolean call(String input, Integer index)
+                {
                     return index < 2;
                 }
             }));

rxjava-core/src/main/java/rx/subjects/Subject.java renamed to rxjava-core/src/main/java/rx/subjects/PublishSubject.java

@@ -19,8 +19,8 @@
 import rx.util.functions.Action1;
 import rx.util.functions.Func1;
 
-public class Subject<T> extends Observable<T> implements Observer<T> {
-    public static <T> Subject<T> create() {
+public class PublishSubject<T> extends Observable<T> implements Observer<T> {
+    public static <T> PublishSubject<T> create() {
         final ConcurrentHashMap<Subscription, Observer<T>> observers = new ConcurrentHashMap<Subscription, Observer<T>>();
 
         Func1<Observer<T>, Subscription> onSubscribe = new Func1<Observer<T>, Subscription>() {
@@ -42,12 +42,12 @@ public void unsubscribe() {
             }
         };
 
-        return new Subject<T>(onSubscribe, observers);
+        return new PublishSubject<T>(onSubscribe, observers);
     }
 
     private final ConcurrentHashMap<Subscription, Observer<T>> observers;
 
-    protected Subject(Func1<Observer<T>, Subscription> onSubscribe, ConcurrentHashMap<Subscription, Observer<T>> observers) {
+    protected PublishSubject(Func1<Observer<T>, Subscription> onSubscribe, ConcurrentHashMap<Subscription, Observer<T>> observers) {
         super(onSubscribe);
         this.observers = observers;
     }
@@ -76,10 +76,10 @@ public void onNext(T args) {
     public static class UnitTest {
         @Test
         public void test() {
-            Subject<Integer> subject = Subject.create();
+            PublishSubject<Integer> publishSubject = PublishSubject.create();
             final AtomicReference<List<Notification<String>>> actualRef = new AtomicReference<List<Notification<String>>>();
 
-            Observable<List<Notification<Integer>>> wNotificationsList = subject.materialize().toList();
+            Observable<List<Notification<Integer>>> wNotificationsList = publishSubject.materialize().toList();
             wNotificationsList.subscribe(new Action1<List<Notification<String>>>() {
                 @Override
                 public void call(List<Notification<String>> actual) {
@@ -108,10 +108,10 @@ public void unsubscribe() {
                         }
                     };
                 }
-            }).subscribe(subject);
-            // the subject has received an onComplete from the first subscribe because
+            }).subscribe(publishSubject);
+            // the publishSubject has received an onComplete from the first subscribe because
             // it is synchronous and the next subscribe won't do anything.
-            Observable.toObservable(-1, -2, -3).subscribe(subject);
+            Observable.toObservable(-1, -2, -3).subscribe(publishSubject);
 
             List<Notification<Integer>> expected = new ArrayList<Notification<Integer>>();
             expected.add(new Notification<Integer>(-1));