Merge pull request #2476 from akarnokd/WindowFixes0121

Fixed off-by-one error and value-drop in the window operator.

Author: akarnokd

src/main/java/rx/internal/operators/OperatorWindowWithTime.java

@@ -33,8 +33,8 @@
 
 /**
  * Creates windows of values into the source sequence with timed window creation, length and size bounds.
- * If timespan == timeshift, windows are non-overlapping but may not be continuous if size number of items were already
- * emitted. If more items arrive after the window has reached its size bound, those items are dropped.
+ * If timespan == timeshift, windows are non-overlapping but always continuous, i.e., when the size bound is reached, a new
+ * window is opened.
  *
  * <p>Note that this conforms the Rx.NET behavior, but does not match former RxJava
  * behavior, which operated as a regular buffer and mapped its lists to Observables.</p>
@@ -205,17 +205,17 @@ void replaceSubject() {
         }
         void emitValue(T t) {
             State<T> s = state;
-            
-            if (s.consumer != null) {
-                s.consumer.onNext(t);
-                if (s.count == size) {
-                    s.consumer.onCompleted();
-                    s = s.clear();
-                } else {
-                    s = s.next();
-                }
+            if (s.consumer == null) {
+                replaceSubject();
+                s = state;
+            }
+            s.consumer.onNext(t);
+            if (s.count == size - 1) {
+                s.consumer.onCompleted();
+                s = s.clear();
+            } else {
+                s = s.next();
             }
-            
             state = s;
         }
 

src/test/java/rx/internal/operators/OperatorWindowWithTimeTest.java

@@ -15,22 +15,17 @@
  */
 package rx.internal.operators;
 
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.fail;
+import static org.junit.Assert.*;
 
-import java.util.ArrayList;
-import java.util.List;
+import java.util.*;
 import java.util.concurrent.TimeUnit;
 
-import org.junit.Before;
-import org.junit.Test;
+import org.junit.*;
 
+import rx.*;
 import rx.Observable;
 import rx.Observer;
-import rx.Scheduler;
-import rx.Subscriber;
-import rx.functions.Action0;
-import rx.functions.Action1;
+import rx.functions.*;
 import rx.schedulers.TestScheduler;
 
 public class OperatorWindowWithTimeTest {
@@ -132,14 +127,14 @@ public void call() {
         }, delay, TimeUnit.MILLISECONDS);
     }
 
-    private Action1<Observable<String>> observeWindow(final List<String> list, final List<List<String>> lists) {
-        return new Action1<Observable<String>>() {
+    private <T> Action1<Observable<T>> observeWindow(final List<T> list, final List<List<T>> lists) {
+        return new Action1<Observable<T>>() {
             @Override
-            public void call(Observable<String> stringObservable) {
-                stringObservable.subscribe(new Observer<String>() {
+            public void call(Observable<T> stringObservable) {
+                stringObservable.subscribe(new Observer<T>() {
                     @Override
                     public void onCompleted() {
-                        lists.add(new ArrayList<String>(list));
+                        lists.add(new ArrayList<T>(list));
                         list.clear();
                     }
 
@@ -149,11 +144,31 @@ public void onError(Throwable e) {
                     }
 
                     @Override
-                    public void onNext(String args) {
+                    public void onNext(T args) {
                         list.add(args);
                     }
                 });
             }
         };
     }
-}
+    @Test
+    public void testExactWindowSize() {
+        Observable<Observable<Integer>> source = Observable.range(1, 10).window(1, TimeUnit.MINUTES, 3, scheduler);
+        
+        final List<Integer> list = new ArrayList<Integer>();
+        final List<List<Integer>> lists = new ArrayList<List<Integer>>();
+        
+        source.subscribe(observeWindow(list, lists));
+        
+        assertEquals(4, lists.size());
+        assertEquals(3, lists.get(0).size());
+        assertEquals(Arrays.asList(1, 2, 3), lists.get(0));
+        assertEquals(3, lists.get(1).size());
+        assertEquals(Arrays.asList(4, 5, 6), lists.get(1));
+        assertEquals(3, lists.get(2).size());
+        assertEquals(Arrays.asList(7, 8, 9), lists.get(2));
+        assertEquals(1, lists.get(3).size());
+        assertEquals(Arrays.asList(10), lists.get(3));
+    }
+    
+}