Timestamp operation

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

@@ -64,6 +64,7 @@
 import rx.operators.OperationTakeLast;
 import rx.operators.OperationTakeUntil;
 import rx.operators.OperationTakeWhile;
+import rx.operators.OperationTimestamp;
 import rx.operators.OperationToFuture;
 import rx.operators.OperationToIterator;
 import rx.operators.OperationToObservableFuture;
@@ -81,6 +82,7 @@
 import rx.util.AtomicObservableSubscription;
 import rx.util.AtomicObserver;
 import rx.util.Range;
+import rx.util.Timestamped;
 import rx.util.functions.Action0;
 import rx.util.functions.Action1;
 import rx.util.functions.Func0;
@@ -252,13 +254,15 @@ public Subscription subscribe(final Map<String, Object> callbacks) {
          */
         return protectivelyWrapAndSubscribe(new Observer() {
 
+            @Override
             public void onCompleted() {
                 Object onComplete = callbacks.get("onCompleted");
                 if (onComplete != null) {
                     Functions.from(onComplete).call();
                 }
             }
 
+            @Override
             public void onError(Exception e) {
                 handleError(e);
                 Object onError = callbacks.get("onError");
@@ -267,6 +271,7 @@ public void onError(Exception e) {
                 }
             }
 
+            @Override
             public void onNext(Object args) {
                 onNext.call(args);
             }
@@ -298,15 +303,18 @@ public Subscription subscribe(final Object o) {
          */
         return protectivelyWrapAndSubscribe(new Observer() {
 
+            @Override
             public void onCompleted() {
                 // do nothing
             }
 
+            @Override
             public void onError(Exception e) {
                 handleError(e);
                 // no callback defined
             }
 
+            @Override
             public void onNext(Object args) {
                 onNext.call(args);
             }
@@ -327,15 +335,18 @@ public Subscription subscribe(final Action1<T> onNext) {
          */
         return protectivelyWrapAndSubscribe(new Observer<T>() {
 
+            @Override
             public void onCompleted() {
                 // do nothing
             }
 
+            @Override
             public void onError(Exception e) {
                 handleError(e);
                 // no callback defined
             }
 
+            @Override
             public void onNext(T args) {
                 if (onNext == null) {
                     throw new RuntimeException("onNext must be implemented");
@@ -365,17 +376,20 @@ public Subscription subscribe(final Object onNext, final Object onError) {
          */
         return protectivelyWrapAndSubscribe(new Observer() {
 
+            @Override
             public void onCompleted() {
                 // do nothing
             }
 
+            @Override
             public void onError(Exception e) {
                 handleError(e);
                 if (onError != null) {
                     Functions.from(onError).call(e);
                 }
             }
 
+            @Override
             public void onNext(Object args) {
                 onNextFunction.call(args);
             }
@@ -396,17 +410,20 @@ public Subscription subscribe(final Action1<T> onNext, final Action1<Exception>
          */
         return protectivelyWrapAndSubscribe(new Observer<T>() {
 
+            @Override
             public void onCompleted() {
                 // do nothing
             }
 
+            @Override
             public void onError(Exception e) {
                 handleError(e);
                 if (onError != null) {
                     onError.call(e);
                 }
             }
 
+            @Override
             public void onNext(T args) {
                 if (onNext == null) {
                     throw new RuntimeException("onNext must be implemented");
@@ -436,19 +453,22 @@ public Subscription subscribe(final Object onNext, final Object onError, final O
          */
         return protectivelyWrapAndSubscribe(new Observer() {
 
+            @Override
             public void onCompleted() {
                 if (onComplete != null) {
                     Functions.from(onComplete).call();
                 }
             }
 
+            @Override
             public void onError(Exception e) {
                 handleError(e);
                 if (onError != null) {
                     Functions.from(onError).call(e);
                 }
             }
 
+            @Override
             public void onNext(Object args) {
                 onNextFunction.call(args);
             }
@@ -469,17 +489,20 @@ public Subscription subscribe(final Action1<T> onNext, final Action1<Exception>
          */
         return protectivelyWrapAndSubscribe(new Observer<T>() {
 
+            @Override
             public void onCompleted() {
                 onComplete.call();
             }
 
+            @Override
             public void onError(Exception e) {
                 handleError(e);
                 if (onError != null) {
                     onError.call(e);
                 }
             }
 
+            @Override
             public void onNext(T args) {
                 if (onNext == null) {
                     throw new RuntimeException("onNext must be implemented");
@@ -516,10 +539,12 @@ public void forEach(final Action1<T> onNext) {
          * See https://github.com/Netflix/RxJava/issues/216 for discussion on "Guideline 6.4: Protect calls to user code from within an operator"
          */
         protectivelyWrapAndSubscribe(new Observer<T>() {
+            @Override
             public void onCompleted() {
                 latch.countDown();
             }
 
+            @Override
             public void onError(Exception e) {
                 /*
                  * If we receive an onError event we set the reference on the outer thread
@@ -531,6 +556,7 @@ public void onError(Exception e) {
                 latch.countDown();
             }
 
+            @Override
             public void onNext(T args) {
                 onNext.call(args);
             }
@@ -582,6 +608,7 @@ public void forEach(final Object o) {
 
         forEach(new Action1() {
 
+            @Override
             public void call(Object args) {
                 onNext.call(args);
             }
@@ -2067,6 +2094,14 @@ public Boolean call(T t, Integer integer)
         }));
     }
 
+    /**
+     * Adds a timestamp to each item emitted by this observable.
+     * @return An observable sequence of timestamped items.
+     */
+    public Observable<Timestamped<T>> timestamp() {
+        return create(OperationTimestamp.timestamp(this));
+    }
+
     /**
      * Return a Future representing a single value of the Observable.
      * <p>
@@ -2743,6 +2778,7 @@ public Observable<T> filter(final Object callback) {
         final FuncN _f = Functions.from(callback);
         return filter(this, new Func1<T, Boolean>() {
 
+            @Override
             public Boolean call(T t1) {
                 return (Boolean) _f.call(t1);
             }
@@ -2913,6 +2949,7 @@ public <R> Observable<R> map(final Object callback) {
         final FuncN _f = Functions.from(callback);
         return map(this, new Func1<T, R>() {
 
+            @Override
             @SuppressWarnings("unchecked")
             public R call(T t1) {
                 return (R) _f.call(t1);
@@ -2963,6 +3000,7 @@ public <R> Observable<R> mapMany(final Object callback) {
         final FuncN _f = Functions.from(callback);
         return mapMany(this, new Func1<T, Observable<R>>() {
 
+            @Override
             @SuppressWarnings("unchecked")
             public Observable<R> call(T t1) {
                 return (Observable<R>) _f.call(t1);
@@ -3071,6 +3109,7 @@ public Observable<T> onErrorResumeNext(final Object resumeFunction) {
         final FuncN _f = Functions.from(resumeFunction);
         return onErrorResumeNext(this, new Func1<Exception, Observable<T>>() {
 
+            @Override
             @SuppressWarnings("unchecked")
             public Observable<T> call(Exception e) {
                 return (Observable<T>) _f.call(e);
@@ -3152,6 +3191,7 @@ public Observable<T> onErrorReturn(final Object resumeFunction) {
         final FuncN _f = Functions.from(resumeFunction);
         return onErrorReturn(this, new Func1<Exception, T>() {
 
+            @Override
             @SuppressWarnings("unchecked")
             public T call(Exception e) {
                 return (T) _f.call(e);

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

@@ -0,0 +1,43 @@
+/**
+ * Copyright 2013 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;
+import rx.Observer;
+import rx.Subscription;
+import rx.util.Timestamped;
+import rx.util.functions.Func1;
+
+public final class OperationTimestamp {
+
+    /**
+     * Accepts a sequence and adds timestamps to each item in it.
+     * 
+     * @param sequence
+     *            the input sequence.
+     * @param <T>
+     *            the type of the input sequence.
+     * @return a sequence of timestamped values created by adding timestamps to each item in the input sequence.
+     */
+    public static <T> Func1<Observer<Timestamped<T>>, Subscription> timestamp(Observable<T> sequence) {
+        return OperationMap.map(sequence, new Func1<T, Timestamped<T>>() {
+            @Override
+            public Timestamped<T> call(T value) {
+                return new Timestamped<T>(System.currentTimeMillis(), value);
+            }
+        });
+    }
+}

rxjava-core/src/main/java/rx/util/Timestamped.java

@@ -0,0 +1,73 @@
+/**
+ * Copyright 2013 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.util;
+
+/**
+ * Composite class that takes a value and a timestamp and wraps them.   
+ */
+public final class Timestamped<T> {
+    private final long timestampMillis;
+    private final T value;
+
+    public Timestamped(long timestampMillis, T value) {
+        this.value = value;
+        this.timestampMillis = timestampMillis;
+    }
+
+    public long getTimestampMillis() {
+        return timestampMillis;
+    }
+
+    public T getValue() {
+        return value;
+    }
+
+    @Override
+    public boolean equals(Object obj) {
+        if (this == obj) {
+            return true;
+        }
+        if (!(obj instanceof Timestamped)) {
+            return false;
+        }
+        Timestamped<?> other = (Timestamped<?>) obj;
+        if (timestampMillis != other.timestampMillis) {
+            return false;
+        }
+        if (value == null) {
+            if (other.value != null) {
+                return false;
+            }
+        } else if (!value.equals(other.value)) {
+            return false;
+        }
+        return true;
+    }
+
+    @Override
+    public int hashCode() {
+        final int prime = 31;
+        int result = 1;
+        result = prime * result + (int) (timestampMillis ^ (timestampMillis));
+        result = prime * result + ((value == null) ? 0 : value.hashCode());
+        return result;
+    }
+
+    @Override
+    public String toString() {
+        return String.format("Timestamped(timestampMillis = %d, value = %s)", timestampMillis, value.toString());
+    }
+}