Bind implementation of Map, Cast, Timestamp

Author: benjchristensen

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

@@ -39,7 +39,7 @@
 import rx.operators.OperationAverage;
 import rx.operators.OperationBuffer;
 import rx.operators.OperationCache;
-import rx.operators.OperationCast;
+import rx.operators.OperatorCast;
 import rx.operators.OperationCombineLatest;
 import rx.operators.OperationConcat;
 import rx.operators.OperationDebounce;
@@ -60,7 +60,7 @@
 import rx.operators.OperationInterval;
 import rx.operators.OperationJoin;
 import rx.operators.OperationJoinPatterns;
-import rx.operators.OperationMap;
+import rx.operators.OperatorMap;
 import rx.operators.OperationMaterialize;
 import rx.operators.OperationMerge;
 import rx.operators.OperationMergeDelayError;
@@ -95,7 +95,7 @@
 import rx.operators.OperationTimeInterval;
 import rx.operators.OperationTimeout;
 import rx.operators.OperationTimer;
-import rx.operators.OperationTimestamp;
+import rx.operators.OperatorTimestamp;
 import rx.operators.OperationToMap;
 import rx.operators.OperationToMultimap;
 import rx.operators.OperationToObservableFuture;
@@ -4043,7 +4043,7 @@ public final Observable<T> cache() {
      * @see <a href="http://msdn.microsoft.com/en-us/library/hh211842.aspx">MSDN: Observable.Cast</a>
      */
     public final <R> Observable<R> cast(final Class<R> klass) {
-        return create(OperationCast.cast(this, klass));
+        return bind(new OperatorCast<T, R>(klass));
     }
 
     /**
@@ -5061,7 +5061,7 @@ public final Long call(Long t1, T t2) {
      * @see <a href="http://msdn.microsoft.com/en-us/library/hh244306.aspx">MSDN: Observable.Select</a>
      */
     public final <R> Observable<R> map(Func1<? super T, ? extends R> func) {
-        return create(OperationMap.map(this, func));
+        return bind(new OperatorMap<T, R>(func));
     }
 
     /**
@@ -5089,6 +5089,7 @@ public final <R> Observable<R> mapMany(Func1<? super T, ? extends Observable<? e
     }
 
     /**
+<<<<<<< HEAD
      * Returns an Observable that applies the specified function to each item emitted by an
      * Observable and emits the results of these function applications.
      * <p>
@@ -5111,6 +5112,10 @@ public final <R> Observable<R> mapWithIndex(Func2<? super T, Integer, ? extends
     /**
      * Turns all of the emissions and notifications from a source Observable into emissions marked
      * with their original types within {@link Notification} objects.
+=======
+     * Turns all of the emissions and notifications from a source Observable
+     * into emissions marked with their original types within {@link Notification} objects.
+>>>>>>> Bind implementation of Map, Cast, Timestamp
      * <p>
      * <img width="640" src="https://raw.github.com/wiki/Netflix/RxJava/images/rx-operators/materialize.png">
      * 
@@ -8157,7 +8162,7 @@ public final Observable<T> timeout(long timeout, TimeUnit timeUnit, Scheduler sc
      * @see <a href="http://msdn.microsoft.com/en-us/library/hh229003.aspx">MSDN: Observable.Timestamp</a>
      */
     public final Observable<Timestamped<T>> timestamp() {
-        return create(OperationTimestamp.timestamp(this));
+        return timestamp(Schedulers.immediate());
     }
 
     /**
@@ -8174,7 +8179,7 @@ public final Observable<Timestamped<T>> timestamp() {
      * @see <a href="http://msdn.microsoft.com/en-us/library/hh229003.aspx">MSDN: Observable.Timestamp</a>
      */
     public final Observable<Timestamped<T>> timestamp(Scheduler scheduler) {
-        return create(OperationTimestamp.timestamp(this, scheduler));
+        return bind(new OperatorTimestamp<T>(scheduler));
     }
 
     /**

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

@@ -0,0 +1,53 @@
+/**
+ * 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.Observable.OperatorSubscription;
+import rx.Observer;
+import rx.util.functions.Func2;
+
+/**
+ * Converts the elements of an observable sequence to the specified type.
+ */
+public class OperatorCast<T, R> implements Func2<Observer<? super R>, OperatorSubscription, Observer<? super T>> {
+
+    private final Class<R> castClass;
+
+    public OperatorCast(Class<R> castClass) {
+        this.castClass = castClass;
+    }
+
+    @Override
+    public Observer<? super T> call(final Observer<? super R> o, OperatorSubscription os) {
+        return new Observer<T>() {
+
+            @Override
+            public void onCompleted() {
+                o.onCompleted();
+            }
+
+            @Override
+            public void onError(Throwable e) {
+                o.onError(e);
+            }
+
+            @Override
+            public void onNext(T t) {
+                o.onNext(castClass.cast(t));
+            }
+        };
+    }
+}

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

@@ -0,0 +1,63 @@
+/**
+ * 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.Observable.OperatorSubscription;
+import rx.Observer;
+import rx.util.functions.Func1;
+import rx.util.functions.Func2;
+
+/**
+ * Applies a function of your choosing to every item emitted by an Observable, and returns this
+ * transformation as a new Observable.
+ * <p>
+ * <img width="640" src="https://github.com/Netflix/RxJava/wiki/images/rx-operators/map.png">
+ */
+public final class OperatorMap<T, R> implements Func2<Observer<? super R>, OperatorSubscription, Observer<? super T>> {
+
+    private final Func1<? super T, ? extends R> transformer;
+
+    public OperatorMap(Func1<? super T, ? extends R> transformer) {
+        this.transformer = transformer;
+    }
+
+    @Override
+    public Observer<? super T> call(final Observer<? super R> o, OperatorSubscription os) {
+        return new Observer<T>() {
+
+            @Override
+            public void onCompleted() {
+                o.onCompleted();
+            }
+
+            @Override
+            public void onError(Throwable e) {
+                o.onError(e);
+            }
+
+            @Override
+            public void onNext(T t) {
+                try {
+                    o.onNext(transformer.call(t));
+                } catch (Throwable e) {
+                    onError(e);
+                }
+            }
+
+        };
+    }
+
+}