Merge pull request #1177 from akarnokd/GroupByUntilTimeGap

GroupByUntil to use BufferUntilSubscriber

Author: benjchristensen

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

@@ -19,11 +19,11 @@
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicReference;
 
-import rx.Observable;
 import rx.Observer;
 import rx.Subscriber;
 import rx.functions.Action0;
 import rx.observers.Subscribers;
+import rx.subjects.Subject;
 import rx.subscriptions.Subscriptions;
 
 /**
@@ -44,7 +44,7 @@
  *
  * @param <T>
  */
-public class BufferUntilSubscriber<T> extends Observable<T> implements Observer<T> {
+public class BufferUntilSubscriber<T> extends Subject<T, T> {
 
     public static <T> BufferUntilSubscriber<T> create() {
         State<T> state = new State<T>();
@@ -122,7 +122,7 @@ public void onNext(T t) {
      * It will then immediately swap itself out for the actual (after a single notification), but since this is now
      * being done on the same producer thread no further buffering will occur.
      */
-    private static class PassThruObserver<T> extends Subscriber<T> {
+    private static final class PassThruObserver<T> extends Subscriber<T> {
 
         private final Observer<? super T> actual;
         // this assumes single threaded synchronous notifications (the Rx contract for a single Observer)
@@ -166,7 +166,7 @@ private void drainIfNeededAndSwitchToActual() {
 
     }
 
-    private static class BufferedObserver<T> extends Subscriber<T> {
+    private static final class BufferedObserver<T> extends Subscriber<T> {
         private final ConcurrentLinkedQueue<Object> buffer = new ConcurrentLinkedQueue<Object>();
         private final NotificationLite<T> nl = NotificationLite.instance();
 

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

@@ -197,7 +197,7 @@ public void onCompleted() {
     public static final class GroupSubject<K, R> extends Subscriber<R> {
 
         static <K, R> GroupSubject<K, R> create(K key) {
-            PublishSubject<R> publish = PublishSubject.create();
+            Subject<R, R> publish = BufferUntilSubscriber.create();
             return new GroupSubject<K, R>(key, publish);
         }
 

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

@@ -31,12 +31,15 @@
 import org.junit.Before;
 import org.junit.Test;
 import org.mockito.InOrder;
+import static org.mockito.Matchers.anyInt;
 import org.mockito.Mock;
+import static org.mockito.Mockito.mock;
 import org.mockito.MockitoAnnotations;
 
 import rx.Observable;
 import rx.Observer;
 import rx.Subscriber;
+import rx.exceptions.TestException;
 import rx.functions.Action1;
 import rx.functions.Func1;
 import rx.functions.Functions;
@@ -283,14 +286,44 @@ public void call(GroupedObservable<Integer, Integer> t1) {
 
         inner.get().subscribe(observer);
 
-        verify(observer, times(1)).onCompleted();
+        verify(observer).onNext(0);
+        verify(observer).onCompleted();
         verify(observer, never()).onError(any(Throwable.class));
-        verify(observer, never()).onNext(any());
+    }
+
+    @Test
+    public void innerEscapeCompletedTwice() {
+        Observable<Integer> source = Observable.from(0);
+
+        final AtomicReference<GroupedObservable<Integer, Integer>> inner = new AtomicReference<GroupedObservable<Integer, Integer>>();
+
+        Func1<GroupedObservable<Integer, Integer>, Observable<Object>> duration = just(Observable.never());
+
+        Observable<GroupedObservable<Integer, Integer>> m = source.groupByUntil(identity, dbl, duration);
+
+        m.subscribe(new Action1<GroupedObservable<Integer, Integer>>() {
+            @Override
+            public void call(GroupedObservable<Integer, Integer> t1) {
+                inner.set(t1);
+            }
+        });
+
+        inner.get().subscribe(observer);
+        
+        @SuppressWarnings("unchecked")
+        Observer<Integer> o2 = mock(Observer.class);
+        
+        inner.get().subscribe(o2);
+
+        verify(o2, never()).onCompleted();
+        verify(o2, never()).onNext(anyInt());
+        verify(o2).onError(any(IllegalStateException.class));
     }
 
     @Test
     public void innerEscapeError() {
-        Observable<Integer> source = Observable.concat(Observable.from(0), Observable.<Integer> error(new RuntimeException("Forced failure")));
+        Observable<Integer> source = Observable.concat(Observable.from(0), Observable.<Integer> error(
+                new TestException("Forced failure")));
 
         final AtomicReference<GroupedObservable<Integer, Integer>> inner = new AtomicReference<GroupedObservable<Integer, Integer>>();
 
@@ -316,8 +349,47 @@ public void onCompleted() {
 
         inner.get().subscribe(observer);
 
-        verify(observer, times(1)).onError(any(Throwable.class));
+        verify(observer).onNext(0);
+        verify(observer).onError(any(TestException.class));
         verify(observer, never()).onCompleted();
-        verify(observer, never()).onNext(any());
+    }
+    
+    @Test
+    public void innerEscapeErrorTwice() {
+        Observable<Integer> source = Observable.concat(Observable.from(0), Observable.<Integer> error(
+                new TestException("Forced failure")));
+
+        final AtomicReference<GroupedObservable<Integer, Integer>> inner = new AtomicReference<GroupedObservable<Integer, Integer>>();
+
+        Func1<GroupedObservable<Integer, Integer>, Observable<Object>> duration = just(Observable.never());
+
+        Observable<GroupedObservable<Integer, Integer>> m = source.groupByUntil(identity, dbl, duration);
+
+        m.subscribe(new Subscriber<GroupedObservable<Integer, Integer>>() {
+            @Override
+            public void onNext(GroupedObservable<Integer, Integer> t1) {
+                inner.set(t1);
+            }
+
+            @Override
+            public void onError(Throwable e) {
+            }
+
+            @Override
+            public void onCompleted() {
+            }
+
+        });
+
+        inner.get().subscribe(observer);
+
+        @SuppressWarnings("unchecked")
+        Observer<Integer> o2 = mock(Observer.class);
+        
+        inner.get().subscribe(o2);
+
+        verify(o2, never()).onCompleted();
+        verify(o2, never()).onNext(anyInt());
+        verify(o2).onError(any(IllegalStateException.class));
     }
 }