OperatorDematerialize

Author: akarnokd

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

@@ -55,7 +55,6 @@
 import rx.operators.OperationDefaultIfEmpty;
 import rx.operators.OperationDefer;
 import rx.operators.OperationDelay;
-import rx.operators.OperationDematerialize;
 import rx.operators.OperationDistinct;
 import rx.operators.OperationDistinctUntilChanged;
 import rx.operators.OperationFinally;
@@ -95,6 +94,7 @@
 import rx.operators.OperatorAsObservable;
 import rx.operators.OperatorCache;
 import rx.operators.OperatorCast;
+import rx.operators.OperatorDematerialize;
 import rx.operators.OperatorDoOnEach;
 import rx.operators.OperatorElementAt;
 import rx.operators.OperatorFilter;
@@ -3590,9 +3590,9 @@ public final Observable<T> delaySubscription(long delay, TimeUnit unit, Schedule
      * @see <a href="https://github.com/Netflix/RxJava/wiki/Observable-Utility-Operators#wiki-dematerialize">RxJava Wiki: dematerialize()</a>
      * @see <a href="http://msdn.microsoft.com/en-us/library/hh229047.aspx">MSDN: Observable.dematerialize</a>
      */
-    @SuppressWarnings("unchecked")
+    @SuppressWarnings({"unchecked", "rawtypes"})
     public final <T2> Observable<T2> dematerialize() {
-        return create(OperationDematerialize.dematerialize((Observable<? extends Notification<? extends T2>>) this));
+        return lift(new OperatorDematerialize());
     }
 
     /**

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

@@ -0,0 +1,76 @@
+/**
+ * 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 rx.Notification;
+import rx.Observable.Operator;
+import rx.Subscriber;
+
+/**
+ * Reverses the effect of {@link OperatorMaterialize} by transforming the Notification objects
+ * emitted by a source Observable into the items or notifications they represent.
+ * <p>
+ * <img width="640" src="https://github.com/Netflix/RxJava/wiki/images/rx-operators/dematerialize.png">
+ * <p>
+ * See <a href="http://msdn.microsoft.com/en-us/library/hh229047(v=vs.103).aspx">here</a> for the
+ * Microsoft Rx equivalent.
+ * 
+ * @param <T> the wrapped value type
+ */
+public final class OperatorDematerialize<T> implements Operator<T, Notification<T>> {
+
+    @Override
+    public Subscriber<? super Notification<T>> call(final Subscriber<? super T> child) {
+        return new Subscriber<Notification<T>>(child) {
+            /** Do not send two onCompleted events. */
+            boolean terminated;
+            @Override
+            public void onNext(Notification<T> t) {
+                switch (t.getKind()) {
+                case OnNext:
+                    if (!terminated) {
+                        child.onNext(t.getValue());
+                    }
+                    break;
+                case OnError:
+                    onError(t.getThrowable());
+                    break;
+                case OnCompleted:
+                    onCompleted();
+                    break;
+                }
+            }
+
+            @Override
+            public void onError(Throwable e) {
+                if (!terminated) {
+                    terminated = true;
+                    child.onError(e);
+                }
+            }
+
+            @Override
+            public void onCompleted() {
+                if (!terminated) {
+                    terminated = true;
+                    child.onCompleted();
+                }
+            }
+            
+        };
+    }
+    
+}

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

@@ -20,16 +20,16 @@
 import static org.mockito.Mockito.never;
 import static org.mockito.Mockito.times;
 import static org.mockito.Mockito.verify;
-import static rx.operators.OperationDematerialize.dematerialize;
 
 import org.junit.Test;
 
 import rx.Notification;
 import rx.Observable;
 import rx.Observer;
+import rx.observers.Subscribers;
 import rx.observers.TestSubscriber;
 
-public class OperationDematerializeTest {
+public class OperatorDematerializeTest {
 
     @Test
     @SuppressWarnings("unchecked")
@@ -51,7 +51,7 @@ public void testDematerialize1() {
     public void testDematerialize2() {
         Throwable exception = new Throwable("test");
         Observable<Integer> observable = Observable.error(exception);
-        Observable<Integer> dematerialize = Observable.create(dematerialize(observable.materialize()));
+        Observable<Integer> dematerialize = observable.materialize().dematerialize();
 
         Observer<Integer> observer = mock(Observer.class);
         dematerialize.subscribe(observer);
@@ -66,7 +66,7 @@ public void testDematerialize2() {
     public void testDematerialize3() {
         Exception exception = new Exception("test");
         Observable<Integer> observable = Observable.error(exception);
-        Observable<Integer> dematerialize = Observable.create(dematerialize(observable.materialize()));
+        Observable<Integer> dematerialize = observable.materialize().dematerialize();
 
         Observer<Integer> observer = mock(Observer.class);
         dematerialize.subscribe(observer);
@@ -106,4 +106,33 @@ public void testCompletePassThru() {
         verify(observer, times(0)).onNext(any(Integer.class));
     }
 
+    @Test
+    public void testHonorsContractWhenCompleted() {
+        Observable<Integer> source = Observable.just(1);
+        
+        Observable<Integer> result = source.materialize().dematerialize();
+        
+        Observer<Integer> o = mock(Observer.class);
+        
+        result.unsafeSubscribe(Subscribers.from(o));
+        
+        verify(o).onNext(1);
+        verify(o).onCompleted();
+        verify(o, never()).onError(any(Throwable.class));
+    }
+    
+    @Test
+    public void testHonorsContractWhenThrows() {
+        Observable<Integer> source = Observable.error(new OperationReduceTest.CustomException());
+        
+        Observable<Integer> result = source.materialize().dematerialize();
+        
+        Observer<Integer> o = mock(Observer.class);
+        
+        result.unsafeSubscribe(Subscribers.from(o));
+        
+        verify(o, never()).onNext(any(Integer.class));
+        verify(o, never()).onCompleted();
+        verify(o).onError(any(OperationReduceTest.CustomException.class));
+    }
 }