Merge pull request #784 from benjchristensen/lift-observer

Lift and Observer+Subscription

Author: benjchristensen

language-adaptors/rxjava-groovy/src/main/java/rx/lang/groovy/GroovyCreateWrapper.java

@@ -16,11 +16,11 @@
 package rx.lang.groovy;
 
 import groovy.lang.Closure;
-import rx.Operator;
+import rx.Observable.OnSubscribe;
+import rx.Observer;
 import rx.Subscription;
-import rx.util.functions.Action1;
 
-public class GroovyCreateWrapper<T> implements Action1<Operator<? super T>> {
+public class GroovyCreateWrapper<T> implements OnSubscribe<T> {
 
     private final Closure<Void> closure;
 
@@ -29,7 +29,7 @@ public GroovyCreateWrapper(Closure<Void> closure) {
     }
 
     @Override
-    public void call(Operator<? super T> op) {
+    public void call(Observer<? super T> op) {
         Object o = closure.call(op);
         /*
          * If the new signature is being used, we will get NULL back.

language-adaptors/rxjava-scala/src/examples/scala/rx/lang/scala/examples/TestSchedulerExample.scala

@@ -17,14 +17,13 @@ package rx.lang.scala.examples
 
 import scala.concurrent.duration.DurationInt
 import scala.language.postfixOps
-
 import org.junit.Test
 import org.mockito.Matchers._
 import org.mockito.Mockito._
 import org.scalatest.junit.JUnitSuite
-
 import rx.lang.scala._
 import rx.lang.scala.schedulers.TestScheduler
+import rx.observers.TestObserver
 
 class TestSchedulerExample extends JUnitSuite {
 
@@ -36,7 +35,7 @@ class TestSchedulerExample extends JUnitSuite {
     val o = Observable.interval(1 second, scheduler)
 
     // Wrap Java Observer in Scala Observer, then subscribe
-    val sub = o.subscribe(Observer(observer))
+    val sub = o.subscribe(Observer(new TestObserver(observer)))
 
     verify(observer, never).onNext(0L)
     verify(observer, never).onCompleted()

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
+  val asJavaObservable: rx.Observable[_ <: T] = asJavaSubject.toObservable()
 
   override val asJavaObserver: rx.Observer[_ >: T] = asJavaSubject
   override def onNext(value: T): Unit = { asJavaObserver.onNext(value)}

rxjava-contrib/rxjava-android/src/test/java/rx/android/observables/AndroidObservableTest.java

@@ -25,15 +25,15 @@
 import org.robolectric.Robolectric;
 import org.robolectric.RobolectricTestRunner;
 import org.robolectric.annotation.Config;
+
 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;
 
 
@@ -66,14 +66,14 @@ public void setup() {
 
     @Test
     public void itSupportsFragmentsFromTheSupportV4Library() {
-        AndroidObservable.fromFragment(supportFragment, Observable.just("success")).subscribe(observer);
+        AndroidObservable.fromFragment(supportFragment, Observable.just("success")).subscribe(new TestObserver<String>(observer));
         verify(observer).onNext("success");
         verify(observer).onCompleted();
     }
 
     @Test
     public void itSupportsNativeFragments() {
-        AndroidObservable.fromFragment(fragment, Observable.just("success")).subscribe(observer);
+        AndroidObservable.fromFragment(fragment, Observable.just("success")).subscribe(new TestObserver<String>(observer));
         verify(observer).onNext("success");
         verify(observer).onCompleted();
     }

rxjava-contrib/rxjava-android/src/test/java/rx/android/operators/OperationObserveFromAndroidComponentTest.java

@@ -42,6 +42,7 @@
 import rx.Observer;
 import rx.Subscription;
 import rx.android.schedulers.AndroidSchedulers;
+import rx.observers.TestObserver;
 import rx.operators.OperationObserveFromAndroidComponent;
 import rx.schedulers.Schedulers;
 import rx.subjects.PublishSubject;
@@ -128,7 +129,7 @@ public void call(Integer i) {
     @Test
     public void itForwardsOnNextOnCompletedSequenceToTargetObserver() {
         Observable<Integer> source = Observable.from(1, 2, 3);
-        OperationObserveFromAndroidComponent.observeFromAndroidComponent(source, mockFragment).subscribe(mockObserver);
+        OperationObserveFromAndroidComponent.observeFromAndroidComponent(source, mockFragment).subscribe(new TestObserver<Integer>(mockObserver));
         verify(mockObserver, times(3)).onNext(anyInt());
         verify(mockObserver).onCompleted();
         verify(mockObserver, never()).onError(any(Exception.class));
@@ -138,7 +139,7 @@ public void itForwardsOnNextOnCompletedSequenceToTargetObserver() {
     public void itForwardsOnErrorToTargetObserver() {
         final Exception exception = new Exception();
         Observable<Integer> source = Observable.error(exception);
-        OperationObserveFromAndroidComponent.observeFromAndroidComponent(source, mockFragment).subscribe(mockObserver);
+        OperationObserveFromAndroidComponent.observeFromAndroidComponent(source, mockFragment).subscribe(new TestObserver<Integer>(mockObserver));
         verify(mockObserver).onError(exception);
         verify(mockObserver, never()).onNext(anyInt());
         verify(mockObserver, never()).onCompleted();
@@ -148,7 +149,7 @@ public void itForwardsOnErrorToTargetObserver() {
     public void itDropsOnNextOnCompletedSequenceIfTargetComponentIsGone() throws Throwable {
         PublishSubject<Integer> source = PublishSubject.create();
 
-        final Observable.OnSubscribeFunc<Integer> operator = newOnSubscribeFragmentInstance(source, mockFragment);
+        final Observable.OnSubscribeFunc<Integer> operator = newOnSubscribeFragmentInstance(source.toObservable(), mockFragment);
         operator.onSubscribe(mockObserver);
 
         source.onNext(1);
@@ -166,7 +167,7 @@ public void itDropsOnNextOnCompletedSequenceIfTargetComponentIsGone() throws Thr
     public void itDropsOnErrorIfTargetComponentIsGone() throws Throwable {
         PublishSubject<Integer> source = PublishSubject.create();
 
-        final Observable.OnSubscribeFunc<Integer> operator = newOnSubscribeFragmentInstance(source, mockFragment);
+        final Observable.OnSubscribeFunc<Integer> operator = newOnSubscribeFragmentInstance(source.toObservable(), mockFragment);
         operator.onSubscribe(mockObserver);
 
         source.onNext(1);
@@ -202,7 +203,7 @@ private void releaseComponentRef(Observable.OnSubscribeFunc<Integer> operator) t
     @Test
     public void itDoesNotForwardOnNextOnCompletedSequenceIfFragmentIsDetached() {
         PublishSubject<Integer> source = PublishSubject.create();
-        OperationObserveFromAndroidComponent.observeFromAndroidComponent(source, mockFragment).subscribe(mockObserver);
+        OperationObserveFromAndroidComponent.observeFromAndroidComponent(source.toObservable(), mockFragment).subscribe(new TestObserver<Integer>(mockObserver));
 
         source.onNext(1);
 
@@ -218,7 +219,7 @@ public void itDoesNotForwardOnNextOnCompletedSequenceIfFragmentIsDetached() {
     @Test
     public void itDoesNotForwardOnErrorIfFragmentIsDetached() {
         PublishSubject<Integer> source = PublishSubject.create();
-        OperationObserveFromAndroidComponent.observeFromAndroidComponent(source, mockFragment).subscribe(mockObserver);
+        OperationObserveFromAndroidComponent.observeFromAndroidComponent(source.toObservable(), mockFragment).subscribe(new TestObserver<Integer>(mockObserver));
 
         source.onNext(1);
 
@@ -244,7 +245,7 @@ public Subscription onSubscribe(Observer<? super Integer> o) {
         });
 
         Subscription sub = OperationObserveFromAndroidComponent.observeFromAndroidComponent(
-                testObservable, mockActivity).subscribe(mockObserver);
+                testObservable, mockActivity).subscribe(new TestObserver<Integer>(mockObserver));
         sub.unsubscribe();
 
         assertTrue(s.isUnsubscribed());

rxjava-contrib/rxjava-android/src/test/java/rx/android/schedulers/HandlerThreadSchedulerTest.java

@@ -15,27 +15,21 @@
  */
 package rx.android.schedulers;
 
-import android.os.Handler;
+import static org.mockito.Matchers.*;
+import static org.mockito.Mockito.*;
+
+import java.util.concurrent.TimeUnit;
 
 import org.junit.Test;
 import org.junit.runner.RunWith;
 import org.mockito.ArgumentCaptor;
-import static org.hamcrest.CoreMatchers.equalTo;
-import static org.junit.Assert.assertNotNull;
-import static org.junit.Assert.assertThat;
 import org.robolectric.RobolectricTestRunner;
 import org.robolectric.annotation.Config;
 
 import rx.Scheduler;
 import rx.Subscription;
-import rx.operators.SafeObservableSubscription;
 import rx.util.functions.Func2;
-
-import java.util.concurrent.TimeUnit;
-
-import static org.mockito.Matchers.eq;
-import static org.mockito.Mockito.mock;
-import static org.mockito.Mockito.verify;
+import android.os.Handler;
 
 @RunWith(RobolectricTestRunner.class)
 @Config(manifest=Config.NONE)

rxjava-contrib/rxjava-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.subscribe(observer));
+                parentSubscription.add(contentSubject.toObservable().subscribe(observer));
                 return parentSubscription;
             }
         });