removed raw Func0s from the chunks

Author: jmhofer

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

@@ -105,7 +105,7 @@ protected static class NonOverlappingChunks<T, C> extends Chunks<T, C> {
 
         private final Object lock = new Object();
 
-        public NonOverlappingChunks(Observer<? super C> observer, Func0 chunkMaker) {
+        public NonOverlappingChunks(Observer<? super C> observer, Func0<? extends Chunk<T, C>> chunkMaker) {
             super(observer, chunkMaker);
         }
 
@@ -134,7 +134,7 @@ public void pushValue(T value) {
      *            <C> The type of object being tracked by the {@link rx.operators.ChunkedOperation.Chunk}
      */
     protected static class OverlappingChunks<T, C> extends Chunks<T, C> {
-        public OverlappingChunks(Observer<? super C> observer, Func0 chunkMaker) {
+        public OverlappingChunks(Observer<? super C> observer, Func0<? extends Chunk<T, C>> chunkMaker) {
             super(observer, chunkMaker);
         }
     }
@@ -158,7 +158,7 @@ protected static class TimeAndSizeBasedChunks<T, C> extends Chunks<T, C> {
         private final TimeUnit unit;
         private final int maxSize;
 
-        public TimeAndSizeBasedChunks(Observer<? super C> observer, Func0 chunkMaker, int maxSize, long maxTime, TimeUnit unit, Scheduler scheduler) {
+        public TimeAndSizeBasedChunks(Observer<? super C> observer, Func0<? extends Chunk<T, C>> chunkMaker, int maxSize, long maxTime, TimeUnit unit, Scheduler scheduler) {
             super(observer, chunkMaker);
             this.maxSize = maxSize;
             this.maxTime = maxTime;
@@ -224,7 +224,7 @@ protected static class TimeBasedChunks<T, C> extends OverlappingChunks<T, C> {
         private final long time;
         private final TimeUnit unit;
 
-        public TimeBasedChunks(Observer<? super C> observer, Func0 chunkMaker, long time, TimeUnit unit, Scheduler scheduler) {
+        public TimeBasedChunks(Observer<? super C> observer, Func0<? extends Chunk<T, C>> chunkMaker, long time, TimeUnit unit, Scheduler scheduler) {
             super(observer, chunkMaker);
             this.time = time;
             this.unit = unit;
@@ -263,7 +263,7 @@ protected static class SizeBasedChunks<T, C> extends Chunks<T, C> {
 
         private final int size;
 
-        public SizeBasedChunks(Observer<? super C> observer, Func0 chunkMaker, int size) {
+        public SizeBasedChunks(Observer<? super C> observer, Func0<? extends Chunk<T, C>> chunkMaker, int size) {
             super(observer, chunkMaker);
             this.size = size;
         }
@@ -295,15 +295,15 @@ protected static class Chunks<T, C> {
 
         private final Queue<Chunk<T, C>> chunks = new ConcurrentLinkedQueue<Chunk<T, C>>();
         private final Observer<? super C> observer;
-        private final Func0<Chunk<T, C>> chunkMaker;
+        private final Func0<? extends Chunk<T, C>> chunkMaker;
 
         /**
          * Constructs a new {@link ChunkedOperation.Chunks} object for the specified {@link rx.Observer}.
          * 
          * @param observer
          *            The {@link rx.Observer} to which this object will emit its internal {@link rx.operators.ChunkedOperation.Chunk} objects to when requested.
          */
-        public Chunks(Observer<? super C> observer, Func0 chunkMaker) {
+        public Chunks(Observer<? super C> observer, Func0<? extends Chunk<T, C>> chunkMaker) {
             this.observer = observer;
             this.chunkMaker = chunkMaker;
         }

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

@@ -76,7 +76,7 @@ public static <T> OnSubscribeFunc<Observable<T>> window(final Observable<T> sour
         return new OnSubscribeFunc<Observable<T>>() {
             @Override
             public Subscription onSubscribe(final Observer<? super Observable<T>> observer) {
-                NonOverlappingChunks<T, Observable<T>> windows = new NonOverlappingChunks<T, Observable<T>>(observer, windowMaker());
+                NonOverlappingChunks<T, Observable<T>> windows = new NonOverlappingChunks<T, Observable<T>>(observer, OperationWindow.<T>windowMaker());
                 ChunkCreator creator = new ObservableBasedSingleChunkCreator<T, Observable<T>>(windows, windowClosingSelector);
                 return source.subscribe(new ChunkObserver<T, Observable<T>>(windows, observer, creator));
             }
@@ -113,7 +113,7 @@ public static <T> OnSubscribeFunc<Observable<T>> window(final Observable<T> sour
         return new OnSubscribeFunc<Observable<T>>() {
             @Override
             public Subscription onSubscribe(final Observer<? super Observable<T>> observer) {
-                OverlappingChunks<T, Observable<T>> windows = new OverlappingChunks<T, Observable<T>>(observer, windowMaker());
+                OverlappingChunks<T, Observable<T>> windows = new OverlappingChunks<T, Observable<T>>(observer, OperationWindow.<T>windowMaker());
                 ChunkCreator creator = new ObservableBasedMultiChunkCreator<T, Observable<T>>(windows, windowOpenings, windowClosingSelector);
                 return source.subscribe(new ChunkObserver<T, Observable<T>>(windows, observer, creator));
             }
@@ -168,7 +168,7 @@ public static <T> OnSubscribeFunc<Observable<T>> window(final Observable<T> sour
         return new OnSubscribeFunc<Observable<T>>() {
             @Override
             public Subscription onSubscribe(final Observer<? super Observable<T>> observer) {
-                Chunks<T, Observable<T>> chunks = new SizeBasedChunks<T, Observable<T>>(observer, windowMaker(), count);
+                Chunks<T, Observable<T>> chunks = new SizeBasedChunks<T, Observable<T>>(observer, OperationWindow.<T>windowMaker(), count);
                 ChunkCreator creator = new SkippingChunkCreator<T, Observable<T>>(chunks, skip);
                 return source.subscribe(new ChunkObserver<T, Observable<T>>(chunks, observer, creator));
             }
@@ -223,7 +223,7 @@ public static <T> OnSubscribeFunc<Observable<T>> window(final Observable<T> sour
         return new OnSubscribeFunc<Observable<T>>() {
             @Override
             public Subscription onSubscribe(final Observer<? super Observable<T>> observer) {
-                NonOverlappingChunks<T, Observable<T>> windows = new NonOverlappingChunks<T, Observable<T>>(observer, windowMaker());
+                NonOverlappingChunks<T, Observable<T>> windows = new NonOverlappingChunks<T, Observable<T>>(observer, OperationWindow.<T>windowMaker());
                 ChunkCreator creator = new TimeBasedChunkCreator<T, Observable<T>>(windows, timespan, unit, scheduler);
                 return source.subscribe(new ChunkObserver<T, Observable<T>>(windows, observer, creator));
             }
@@ -284,7 +284,7 @@ public static <T> OnSubscribeFunc<Observable<T>> window(final Observable<T> sour
         return new OnSubscribeFunc<Observable<T>>() {
             @Override
             public Subscription onSubscribe(final Observer<? super Observable<T>> observer) {
-                Chunks<T, Observable<T>> chunks = new TimeAndSizeBasedChunks<T, Observable<T>>(observer, windowMaker(), count, timespan, unit, scheduler);
+                Chunks<T, Observable<T>> chunks = new TimeAndSizeBasedChunks<T, Observable<T>>(observer, OperationWindow.<T>windowMaker(), count, timespan, unit, scheduler);
                 ChunkCreator creator = new SingleChunkCreator<T, Observable<T>>(chunks);
                 return source.subscribe(new ChunkObserver<T, Observable<T>>(chunks, observer, creator));
             }
@@ -345,7 +345,7 @@ public static <T> OnSubscribeFunc<Observable<T>> window(final Observable<T> sour
         return new OnSubscribeFunc<Observable<T>>() {
             @Override
             public Subscription onSubscribe(final Observer<? super Observable<T>> observer) {
-                OverlappingChunks<T, Observable<T>> windows = new TimeBasedChunks<T, Observable<T>>(observer, windowMaker(), timespan, unit, scheduler);
+                OverlappingChunks<T, Observable<T>> windows = new TimeBasedChunks<T, Observable<T>>(observer, OperationWindow.<T>windowMaker(), timespan, unit, scheduler);
                 ChunkCreator creator = new TimeBasedChunkCreator<T, Observable<T>>(windows, timeshift, unit, scheduler);
                 return source.subscribe(new ChunkObserver<T, Observable<T>>(windows, observer, creator));
             }