Debounce with selector

Author: akarnokd

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

@@ -2266,6 +2266,20 @@ public Observable<T> debounce(long timeout, TimeUnit unit) {
         return create(OperationDebounce.debounce(this, timeout, unit));
     }
 
+    /**
+     * Create an Observable that ignores elements from this observable
+     * sequence which are followed by another value within a computed
+     * debounce duration.
+     * @param <U> the debounce value type (ignored)
+     * @param debounceSelector function to retrieve a sequence indicating the throttle duration for each given element.
+     * @return an Observable that ignores elements from this observable
+     * sequence which are followed by another value within a computed
+     * debounce duration
+     */
+    public <U> Observable<T> debounce(Func1<? super T, ? extends Observable<U>> debounceSelector) {
+        return create(OperationDebounce.debounceSelector(this, debounceSelector));
+    }
+    
     /**
      * Drops items emitted by an Observable that are followed by newer items
      * before a timeout value expires. The timer resets on each emission.

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

@@ -24,6 +24,8 @@
 import rx.Scheduler;
 import rx.Subscription;
 import rx.schedulers.Schedulers;
+import rx.subscriptions.CompositeSubscription;
+import rx.subscriptions.SerialSubscription;
 import rx.util.functions.Action0;
 import rx.util.functions.Func1;
 
@@ -151,4 +153,162 @@ public void call() {
             }
         }
     }
+    
+    /**
+     * Delay the emission via another observable if no new source appears in the meantime.
+     */
+    public static <T, U> OnSubscribeFunc<T> debounceSelector(
+            Observable<? extends T> source,
+            Func1<? super T, ? extends Observable<U>> debounceSelector) {
+        return new DebounceSelector<T, U>(source, debounceSelector);
+    }
+    
+    /**
+     * Delay the emission via another observable if no new source appears in the meantime.
+     */
+    private static final class DebounceSelector<T, U> implements OnSubscribeFunc<T> {
+        final Observable<? extends T> source;
+        final Func1<? super T, ? extends Observable<U>> debounceSelector;
+
+        public DebounceSelector(Observable<? extends T> source, Func1<? super T, ? extends Observable<U>> debounceSelector) {
+            this.source = source;
+            this.debounceSelector = debounceSelector;
+        }
+
+        @Override
+        public Subscription onSubscribe(Observer<? super T> t1) {
+            CompositeSubscription csub = new CompositeSubscription();
+            
+            csub.add(source.subscribe(new SourceObserver<T, U>(t1, debounceSelector, csub)));
+            
+            return csub;
+        }
+        
+        /** Observe the source. */
+        private static final class SourceObserver<T, U> implements Observer<T> {
+            final Observer<? super T> observer;
+            final Func1<? super T, ? extends Observable<U>> debounceSelector;
+            final CompositeSubscription cancel;
+            final SerialSubscription ssub = new SerialSubscription();
+            long index;
+            T value;
+            boolean hasValue;
+            final Object guard;
+            
+            public SourceObserver(
+                    Observer<? super T> observer, 
+                    Func1<? super T, ? extends Observable<U>> debounceSelector, 
+                    CompositeSubscription cancel) {
+                this.observer = observer;
+                this.debounceSelector = debounceSelector;
+                this.cancel = cancel;
+                this.cancel.add(ssub);
+                this.guard = new Object();
+            }
+
+            @Override
+            public void onNext(T args) {
+                Observable<U> o;
+                try {
+                    o = debounceSelector.call(args);
+                } catch (Throwable t) {
+                    synchronized (guard) {
+                        observer.onError(t);
+                    }
+                    cancel.unsubscribe();
+                    return;
+                }
+                long currentIndex;
+                synchronized (guard) {
+                    hasValue = true;
+                    value = args;
+                    currentIndex = ++index;
+                }
+                
+                SerialSubscription osub = new SerialSubscription();
+                ssub.set(osub);
+
+                osub.set(o.subscribe(new DebounceObserver<T, U>(this, osub, args, currentIndex)));
+            }
+
+            @Override
+            public void onError(Throwable e) {
+                ssub.unsubscribe();
+                try {
+                    synchronized (guard) {
+                        observer.onError(e);
+                        hasValue = false;
+                        value = null;
+                        index++;
+                    }
+                } finally {
+                    cancel.unsubscribe();
+                }
+            }
+
+            @Override
+            public void onCompleted() {
+                ssub.unsubscribe();
+                try {
+                    synchronized (guard) {
+                        if (hasValue) {
+                            try {
+                                observer.onNext(value);
+                            } catch (Throwable t) {
+                                observer.onError(t);
+                                return;
+                            }
+                        }
+                        observer.onCompleted();
+                        hasValue = false;
+                        value = null;
+                        index++;
+                    }
+                } finally {
+                    cancel.unsubscribe();
+                }
+            }
+        }
+        /**
+         * The debounce observer.
+         */
+        private static final class DebounceObserver<T, U> implements Observer<U> {
+            final SourceObserver<T, U> parent;
+            final Subscription cancel;
+            final T value;
+            final long currentIndex;
+
+            public DebounceObserver(SourceObserver<T, U> parent, Subscription cancel, T value, long currentIndex) {
+                this.parent = parent;
+                this.cancel = cancel;
+                this.value = value;
+                this.currentIndex = currentIndex;
+            }
+
+            @Override
+            public void onNext(U args) {
+                onCompleted();
+            }
+
+            @Override
+            public void onError(Throwable e) {
+                synchronized (parent.guard) {
+                    parent.observer.onError(e);
+                }
+                parent.cancel.unsubscribe();
+            }
+
+            @Override
+            public void onCompleted() {
+                synchronized (parent.guard) {
+                    if (parent.hasValue && parent.index == currentIndex) {
+                        parent.observer.onNext(value);
+                    }
+                    parent.hasValue = false;
+                }
+                cancel.unsubscribe();
+            }
+            
+        }
+    }
 }

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

@@ -15,7 +15,6 @@
  */
 package rx.operators;
 
-import static org.mockito.Matchers.*;
 import static org.mockito.Mockito.*;
 
 import java.util.concurrent.TimeUnit;
@@ -28,8 +27,10 @@
 import rx.Observer;
 import rx.Subscription;
 import rx.schedulers.TestScheduler;
+import rx.subjects.PublishSubject;
 import rx.subscriptions.Subscriptions;
 import rx.util.functions.Action0;
+import rx.util.functions.Func1;
 
 public class OperationDebounceTest {
 
@@ -158,4 +159,88 @@ public void call() {
     @SuppressWarnings("serial")
     private class TestException extends Exception {
     }
+    
+    @Test
+    public void debounceSelectorNormal1() {
+        PublishSubject<Integer> source = PublishSubject.create();
+        final PublishSubject<Integer> debouncer = PublishSubject.create();
+        Func1<Integer, Observable<Integer>> debounceSel = new Func1<Integer, Observable<Integer>>() {
+
+            @Override
+            public Observable<Integer> call(Integer t1) {
+                return debouncer;
+            }
+        };
+        
+        @SuppressWarnings("unchecked")
+        Observer<Object> o = mock(Observer.class);
+        InOrder inOrder = inOrder(o);
+        
+        source.debounce(debounceSel).subscribe(o);
+        
+        source.onNext(1);
+        debouncer.onNext(1);
+        
+        source.onNext(2);
+        source.onNext(3);
+        source.onNext(4);
+
+        debouncer.onNext(2);
+        
+        source.onNext(5);
+        source.onCompleted();
+        
+        inOrder.verify(o).onNext(1);
+        inOrder.verify(o).onNext(4);
+        inOrder.verify(o).onNext(5);
+        inOrder.verify(o).onCompleted();
+        
+        verify(o, never()).onError(any(Throwable.class));
+    }
+    
+    @Test
+    public void debounceSelectorFuncThrows() {
+        PublishSubject<Integer> source = PublishSubject.create();
+        Func1<Integer, Observable<Integer>> debounceSel = new Func1<Integer, Observable<Integer>>() {
+
+            @Override
+            public Observable<Integer> call(Integer t1) {
+                throw new OperationReduceTest.CustomException();
+            }
+        };
+        
+        @SuppressWarnings("unchecked")
+        Observer<Object> o = mock(Observer.class);
+        
+        source.debounce(debounceSel).subscribe(o);
+        
+        source.onNext(1);
+        
+        verify(o, never()).onNext(any());
+        verify(o, never()).onCompleted();
+        verify(o).onError(any(OperationReduceTest.CustomException.class));
+    }
+    
+    @Test
+    public void debounceSelectorObservableThrows() {
+        PublishSubject<Integer> source = PublishSubject.create();
+        Func1<Integer, Observable<Integer>> debounceSel = new Func1<Integer, Observable<Integer>>() {
+
+            @Override
+            public Observable<Integer> call(Integer t1) {
+                return Observable.error(new OperationReduceTest.CustomException());
+            }
+        };
+        
+        @SuppressWarnings("unchecked")
+        Observer<Object> o = mock(Observer.class);
+        
+        source.debounce(debounceSel).subscribe(o);
+        
+        source.onNext(1);
+        
+        verify(o, never()).onNext(any());
+        verify(o, never()).onCompleted();
+        verify(o).onError(any(OperationReduceTest.CustomException.class));
+    }
 }