Merge pull request #793 from benjchristensen/subscriber-observer

Observer + Subscriber
ReactiveX · Jan 31, 2014 · 025de00 · 025de00
2 parents ef12fe3 + 171ebc0
commit 025de00
Show file tree

Hide file tree

Showing 211 changed files with 1,848 additions and 1,385 deletions.
diff --git a/language-adaptors/rxjava-groovy/src/main/java/rx/lang/groovy/GroovyCreateWrapper.java b/language-adaptors/rxjava-groovy/src/main/java/rx/lang/groovy/GroovyCreateWrapper.java
@@ -17,7 +17,7 @@
 
 import groovy.lang.Closure;
 import rx.Observable.OnSubscribe;
-import rx.Observer;
+import rx.Subscriber;
 import rx.Subscription;
 
 public class GroovyCreateWrapper<T> implements OnSubscribe<T> {
@@ -29,7 +29,7 @@ public GroovyCreateWrapper(Closure<Void> closure) {
     }
 
     @Override
-    public void call(Observer<? super T> op) {
+    public void call(Subscriber<? super T> op) {
         Object o = closure.call(op);
         /*
          * If the new signature is being used, we will get NULL back.

diff --git a/language-adaptors/rxjava-groovy/src/test/groovy/rx/lang/groovy/ObservableTests.groovy b/language-adaptors/rxjava-groovy/src/test/groovy/rx/lang/groovy/ObservableTests.groovy
@@ -15,28 +15,21 @@
  */
 package rx.lang.groovy
 
-import static org.mockito.Matchers.*;
-import static org.mockito.Mockito.*;
-
-import java.util.Arrays;
-import java.util.Collection;
-import java.util.Map;
-
-import org.junit.Before;
-import org.junit.Test;
-import static org.junit.Assert.*;
-
-import org.mockito.Mock;
-import org.mockito.MockitoAnnotations;
-
-import rx.Notification;
-import rx.Observable;
-import rx.Observable.OnSubscribeFunc;
-import rx.Observer;
-import rx.Subscription;
-import rx.observables.GroupedObservable;
-import rx.subscriptions.Subscriptions;
-import rx.util.functions.Func1;
+import static org.junit.Assert.*
+import static org.mockito.Matchers.*
+import static org.mockito.Mockito.*
+
+import org.junit.Before
+import org.junit.Test
+import org.mockito.Mock
+import org.mockito.MockitoAnnotations
+
+import rx.Notification
+import rx.Observable
+import rx.Observer
+import rx.Subscription
+import rx.Observable.OnSubscribeFunc
+import rx.subscriptions.Subscriptions
 
 def class ObservableTests {
 

diff --git a/language-adaptors/rxjava-jruby/src/main/java/rx/lang/jruby/JRubyActionWrapper.java b/language-adaptors/rxjava-jruby/src/main/java/rx/lang/jruby/JRubyActionWrapper.java
@@ -15,11 +15,11 @@
  */
 package rx.lang.jruby;
 
-import org.jruby.RubyProc;
 import org.jruby.Ruby;
+import org.jruby.RubyProc;
+import org.jruby.javasupport.JavaUtil;
 import org.jruby.runtime.ThreadContext;
 import org.jruby.runtime.builtin.IRubyObject;
-import org.jruby.javasupport.JavaUtil;
 
 import rx.util.functions.Action;
 import rx.util.functions.Action0;

diff --git a/language-adaptors/rxjava-jruby/src/main/java/rx/lang/jruby/JRubyFunctionWrapper.java b/language-adaptors/rxjava-jruby/src/main/java/rx/lang/jruby/JRubyFunctionWrapper.java
@@ -15,11 +15,11 @@
  */
 package rx.lang.jruby;
 
-import org.jruby.RubyProc;
 import org.jruby.Ruby;
+import org.jruby.RubyProc;
+import org.jruby.javasupport.JavaUtil;
 import org.jruby.runtime.ThreadContext;
 import org.jruby.runtime.builtin.IRubyObject;
-import org.jruby.javasupport.JavaUtil;
 
 import rx.util.functions.Func0;
 import rx.util.functions.Func1;

diff --git a/language-adaptors/rxjava-scala/src/main/scala/rx/lang/scala/Subject.scala b/language-adaptors/rxjava-scala/src/main/scala/rx/lang/scala/Subject.scala
@@ -21,7 +21,7 @@ package rx.lang.scala
 trait Subject[T] extends Observable[T] with Observer[T] {
   private [scala] val asJavaSubject: rx.subjects.Subject[_ >: T, _<: T]
 
-  val asJavaObservable: rx.Observable[_ <: T] = asJavaSubject.toObservable()
+  val asJavaObservable: rx.Observable[_ <: T] = asJavaSubject
 
   override val asJavaObserver: rx.Observer[_ >: T] = asJavaSubject
   override def onNext(value: T): Unit = { asJavaObserver.onNext(value)}

diff --git a/rxjava-contrib/rxjava-android/src/main/java/rx/android/observables/AndroidObservable.java b/rxjava-contrib/rxjava-android/src/main/java/rx/android/observables/AndroidObservable.java
@@ -16,13 +16,10 @@
 package rx.android.observables;
 
 import rx.Observable;
-import rx.Observer;
 import rx.operators.OperationObserveFromAndroidComponent;
-
 import android.app.Activity;
 import android.app.Fragment;
 import android.os.Build;
-import android.support.v4.app.FragmentActivity;
 
 
 public final class AndroidObservable {

diff --git a/rxjava-contrib/rxjava-android/src/main/java/rx/android/schedulers/AndroidSchedulers.java b/rxjava-contrib/rxjava-android/src/main/java/rx/android/schedulers/AndroidSchedulers.java
@@ -15,9 +15,9 @@
  */
 package rx.android.schedulers;
 
+import rx.Scheduler;
 import android.os.Handler;
 import android.os.Looper;
-import rx.Scheduler;
 
 /**
  * Schedulers that have Android specific functionality

diff --git a/...va-contrib/rxjava-android/src/main/java/rx/android/schedulers/HandlerThreadScheduler.java b/...va-contrib/rxjava-android/src/main/java/rx/android/schedulers/HandlerThreadScheduler.java
@@ -15,14 +15,13 @@
  */
 package rx.android.schedulers;
 
-import android.os.Handler;
+import java.util.concurrent.TimeUnit;
 
 import rx.Scheduler;
 import rx.Subscription;
 import rx.operators.SafeObservableSubscription;
 import rx.util.functions.Func2;
-
-import java.util.concurrent.TimeUnit;
+import android.os.Handler;
 
 /**
  * Schedules actions to run on an Android Handler thread.

diff --git a/...ntrib/rxjava-android/src/main/java/rx/operators/OperationObserveFromAndroidComponent.java b/...ntrib/rxjava-android/src/main/java/rx/operators/OperationObserveFromAndroidComponent.java
@@ -19,17 +19,10 @@
 import rx.Observer;
 import rx.Subscription;
 import rx.android.schedulers.AndroidSchedulers;
-import rx.subjects.PublishSubject;
 import android.app.Activity;
-import android.app.Fragment;
 import android.os.Looper;
 import android.util.Log;
 
-import java.util.concurrent.Callable;
-import java.util.concurrent.Executors;
-import java.util.concurrent.Future;
-import java.util.concurrent.TimeUnit;
-
 public class OperationObserveFromAndroidComponent {
 
     public static <T> Observable<T> observeFromAndroidComponent(Observable<T> source, android.app.Fragment fragment) {

diff --git a/...va-contrib/rxjava-android/src/test/java/rx/android/observables/AndroidObservableTest.java b/...va-contrib/rxjava-android/src/test/java/rx/android/observables/AndroidObservableTest.java
@@ -15,7 +15,7 @@
  */
 package rx.android.observables;
 
-import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.*;
 
 import org.junit.Before;
 import org.junit.Test;
@@ -29,12 +29,9 @@
 import rx.Observable;
 import rx.Observer;
 import rx.observers.TestObserver;
-import rx.operators.OperationObserveFromAndroidComponent;
 import android.app.Activity;
 import android.app.Fragment;
-import android.os.Build;
 import android.support.v4.app.FragmentActivity;
-import rx.android.observables.AndroidObservable;
 
 
 @RunWith(RobolectricTestRunner.class)

diff --git a/...-android/src/test/java/rx/android/operators/OperationObserveFromAndroidComponentTest.java b/...-android/src/test/java/rx/android/operators/OperationObserveFromAndroidComponentTest.java
@@ -43,6 +43,7 @@
 import rx.Subscription;
 import rx.android.schedulers.AndroidSchedulers;
 import rx.observers.TestObserver;
+import rx.observers.TestSubscriber;
 import rx.operators.OperationObserveFromAndroidComponent;
 import rx.schedulers.Schedulers;
 import rx.subjects.PublishSubject;
@@ -149,8 +150,8 @@ public void itForwardsOnErrorToTargetObserver() {
     public void itDropsOnNextOnCompletedSequenceIfTargetComponentIsGone() throws Throwable {
         PublishSubject<Integer> source = PublishSubject.create();
 
-        final Observable.OnSubscribeFunc<Integer> operator = newOnSubscribeFragmentInstance(source.toObservable(), mockFragment);
-        operator.onSubscribe(mockObserver);
+        final Observable.OnSubscribeFunc<Integer> operator = newOnSubscribeFragmentInstance(source, mockFragment);
+        operator.onSubscribe(new TestSubscriber<Integer>(mockObserver));
 
         source.onNext(1);
         releaseComponentRef(operator);
@@ -167,8 +168,8 @@ public void itDropsOnNextOnCompletedSequenceIfTargetComponentIsGone() throws Thr
     public void itDropsOnErrorIfTargetComponentIsGone() throws Throwable {
         PublishSubject<Integer> source = PublishSubject.create();
 
-        final Observable.OnSubscribeFunc<Integer> operator = newOnSubscribeFragmentInstance(source.toObservable(), mockFragment);
-        operator.onSubscribe(mockObserver);
+        final Observable.OnSubscribeFunc<Integer> operator = newOnSubscribeFragmentInstance(source, mockFragment);
+        operator.onSubscribe(new TestSubscriber<Integer>(mockObserver));
 
         source.onNext(1);
         releaseComponentRef(operator);
@@ -203,7 +204,7 @@ private void releaseComponentRef(Observable.OnSubscribeFunc<Integer> operator) t
     @Test
     public void itDoesNotForwardOnNextOnCompletedSequenceIfFragmentIsDetached() {
         PublishSubject<Integer> source = PublishSubject.create();
-        OperationObserveFromAndroidComponent.observeFromAndroidComponent(source.toObservable(), mockFragment).subscribe(new TestObserver<Integer>(mockObserver));
+        OperationObserveFromAndroidComponent.observeFromAndroidComponent(source, mockFragment).subscribe(new TestObserver<Integer>(mockObserver));
 
         source.onNext(1);
 
@@ -219,7 +220,7 @@ public void itDoesNotForwardOnNextOnCompletedSequenceIfFragmentIsDetached() {
     @Test
     public void itDoesNotForwardOnErrorIfFragmentIsDetached() {
         PublishSubject<Integer> source = PublishSubject.create();
-        OperationObserveFromAndroidComponent.observeFromAndroidComponent(source.toObservable(), mockFragment).subscribe(new TestObserver<Integer>(mockObserver));
+        OperationObserveFromAndroidComponent.observeFromAndroidComponent(source, mockFragment).subscribe(new TestObserver<Integer>(mockObserver));
 
         source.onNext(1);
 

diff --git a/...xjava-apache-http/src/main/java/rx/apache/http/consumers/ResponseConsumerEventStream.java b/...xjava-apache-http/src/main/java/rx/apache/http/consumers/ResponseConsumerEventStream.java
@@ -85,7 +85,7 @@ protected void onResponseReceived(HttpResponse response) throws HttpException, I
 
             @Override
             public Subscription onSubscribe(Observer<? super byte[]> observer) {
-                parentSubscription.add(contentSubject.toObservable().subscribe(observer));
+                parentSubscription.add(contentSubject.subscribe(observer));
                 return parentSubscription;
             }
         });

diff --git a/rxjava-contrib/rxjava-async-util/src/main/java/rx/util/async/Async.java b/rxjava-contrib/rxjava-async-util/src/main/java/rx/util/async/Async.java
@@ -18,6 +18,7 @@
 import java.util.concurrent.Callable;
 import java.util.concurrent.Future;
 import java.util.concurrent.FutureTask;
+
 import rx.Observable;
 import rx.Scheduler;
 import rx.schedulers.Schedulers;
@@ -605,7 +606,7 @@ public void call() {
                         subject.onCompleted();
                     }
                 });
-                return subject.toObservable();
+                return subject;
             }
         };
     }
@@ -662,7 +663,7 @@ public void call() {
                         subject.onCompleted();
                     }
                 });
-                return subject.toObservable();
+                return subject;
             }
         };
     }
@@ -721,7 +722,7 @@ public void call() {
                         subject.onCompleted();
                     }
                 });
-                return subject.toObservable();
+                return subject;
             }
         };
     }
@@ -782,7 +783,7 @@ public void call() {
                         subject.onCompleted();
                     }
                 });
-                return subject.toObservable();
+                return subject;
             }
         };
     }
@@ -845,7 +846,7 @@ public void call() {
                         subject.onCompleted();
                     }
                 });
-                return subject.toObservable();
+                return subject;
             }
         };
     }
@@ -910,7 +911,7 @@ public void call() {
                         subject.onCompleted();
                     }
                 });
-                return subject.toObservable();
+                return subject;
             }
         };
     }
@@ -977,7 +978,7 @@ public void call() {
                         subject.onCompleted();
                     }
                 });
-                return subject.toObservable();
+                return subject;
             }
         };
     }
@@ -1046,7 +1047,7 @@ public void call() {
                         subject.onCompleted();
                     }
                 });
-                return subject.toObservable();
+                return subject;
             }
         };
     }
@@ -1117,7 +1118,7 @@ public void call() {
                         subject.onCompleted();
                     }
                 });
-                return subject.toObservable();
+                return subject;
             }
         };
     }
@@ -1190,7 +1191,7 @@ public void call() {
                         subject.onCompleted();
                     }
                 });
-                return subject.toObservable();
+                return subject;
             }
         };
     }
@@ -1243,7 +1244,7 @@ public void call() {
                         subject.onCompleted();
                     }
                 });
-                return subject.toObservable();
+                return subject;
             }
         };
     }

diff --git a/rxjava-contrib/rxjava-async-util/src/main/java/rx/util/async/operators/LatchedObserver.java b/rxjava-contrib/rxjava-async-util/src/main/java/rx/util/async/operators/LatchedObserver.java
@@ -18,6 +18,7 @@
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
+
 import rx.Observer;
 import rx.util.functions.Action0;
 import rx.util.functions.Action1;
@@ -28,7 +29,7 @@
  * a terminal state has been reached.
  * @param <T> the observed value type
  */
-abstract class LatchedObserver<T> extends Observer<T> {
+abstract class LatchedObserver<T> implements Observer<T> {
     /** The CountDownLatch to count-down on a terminal state. */
     protected final CountDownLatch latch;
     /** Contains the error. */

diff --git a/...contrib/rxjava-async-util/src/main/java/rx/util/async/operators/OperationDeferFuture.java b/...contrib/rxjava-async-util/src/main/java/rx/util/async/operators/OperationDeferFuture.java
@@ -16,10 +16,10 @@
 package rx.util.async.operators;
 
 import java.util.concurrent.Future;
+
 import rx.Observable;
 import rx.Scheduler;
 import rx.util.functions.Func0;
-import rx.util.functions.Func1;
 
 /**
  * Defer the execution of a factory method which produces an observable sequence.

diff --git a/...ntrib/rxjava-async-util/src/main/java/rx/util/async/operators/OperationForEachFuture.java b/...ntrib/rxjava-async-util/src/main/java/rx/util/async/operators/OperationForEachFuture.java
@@ -17,6 +17,7 @@
 
 import java.util.concurrent.Callable;
 import java.util.concurrent.FutureTask;
+
 import rx.Observable;
 import rx.Subscription;
 import rx.util.Exceptions;

diff --git a/...rib/rxjava-async-util/src/main/java/rx/util/async/operators/OperationFromFunctionals.java b/...rib/rxjava-async-util/src/main/java/rx/util/async/operators/OperationFromFunctionals.java
@@ -16,6 +16,7 @@
 package rx.util.async.operators;
 
 import java.util.concurrent.Callable;
+
 import rx.Observable.OnSubscribeFunc;
 import rx.Observer;
 import rx.Subscription;