OperatorAsObservable

Author: akarnokd

rxjava-core/src/main/java/rx/Observable.java

@@ -50,7 +50,7 @@
 import rx.operators.OnSubscribeRange;
 import rx.operators.OperationAll;
 import rx.operators.OperationAny;
-import rx.operators.OperationAsObservable;
+import rx.operators.OperatorAsObservable;
 import rx.operators.OperationBuffer;
 import rx.operators.OperationCombineLatest;
 import rx.operators.OperationConcat;
@@ -2952,7 +2952,7 @@ public final Observable<Boolean> all(Func1<? super T, Boolean> predicate) {
      * @return an Observable that hides the identity of this Observable
      */
     public final Observable<T> asObservable() {
-        return create(new OperationAsObservable<T>(this));
+        return lift(new OperatorAsObservable<T>());
     }
 
     /**

rxjava-core/src/main/java/rx/operators/OperationAsObservable.java renamed to rxjava-core/src/main/java/rx/operators/OperatorAsObservable.java

@@ -15,27 +15,20 @@
  */
 package rx.operators;
 
-import rx.Observable;
-import rx.Observable.OnSubscribeFunc;
-import rx.Observer;
-import rx.Subscription;
-import rx.observers.Subscribers;
+import rx.Observable.Operator;
+import rx.Subscriber;
 
 /**
  * Hides the identity of another observable.
  * 
  * @param <T>
  *            the return value type of the wrapped observable.
  */
-public final class OperationAsObservable<T> implements OnSubscribeFunc<T> {
-    private final Observable<? extends T> source;
-
-    public OperationAsObservable(Observable<? extends T> source) {
-        this.source = source;
-    }
+public final class OperatorAsObservable<T> implements Operator<T, T> {
 
     @Override
-    public Subscription onSubscribe(final Observer<? super T> t1) {
-        return source.unsafeSubscribe(Subscribers.from(t1));
+    public Subscriber<? super T> call(Subscriber<? super T> t1) {
+        return t1;
     }
+    
 }

rxjava-core/src/test/java/rx/operators/OperatorAsObservableTest.java

@@ -0,0 +1,65 @@
+/**
+ * Copyright 2014 Netflix, Inc.
+ * 
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ * 
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * 
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package rx.operators;
+
+import static org.junit.Assert.assertFalse;
+import org.junit.Test;
+import static org.mockito.Matchers.any;
+import static org.mockito.Mockito.*;
+import rx.Observable;
+import rx.Observer;
+import rx.subjects.PublishSubject;
+
+public class OperatorAsObservableTest {
+    @Test
+    public void testHiding() {
+        PublishSubject<Integer> src = PublishSubject.create();
+        
+        Observable<Integer> dst = src.asObservable();
+        
+        assertFalse(dst instanceof PublishSubject);
+        
+        Observer<Object> o = mock(Observer.class);
+        
+        dst.subscribe(o);
+        
+        src.onNext(1);
+        src.onCompleted();
+        
+        verify(o).onNext(1);
+        verify(o).onCompleted();
+        verify(o, never()).onError(any(Throwable.class));
+    }
+    @Test
+    public void testHidingError() {
+        PublishSubject<Integer> src = PublishSubject.create();
+        
+        Observable<Integer> dst = src.asObservable();
+        
+        assertFalse(dst instanceof PublishSubject);
+        
+        Observer<Object> o = mock(Observer.class);
+        
+        dst.subscribe(o);
+        
+        src.onError(new OperationReduceTest.CustomException());
+        
+        verify(o, never()).onNext(any());
+        verify(o, never()).onCompleted();
+        verify(o).onError(any(OperationReduceTest.CustomException.class));
+    }
+}