made the public window methods more generic via the basic (lol) super/extends fluff; also simplified api by removing a few useless super definitions (there's no super of Opening and Closing)

Author: jmhofer

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

@@ -1337,7 +1337,7 @@ public Observable<List<T>> buffer(Func0<? extends Observable<? extends Closing>>
      * @return
      *         An {@link Observable} which produces buffers which are created and emitted when the specified {@link Observable}s publish certain objects.
      */
-    public Observable<List<T>> buffer(Observable<? extends Opening> bufferOpenings, Func1<? super Opening, ? extends Observable<? extends Closing>> bufferClosingSelector) {
+    public Observable<List<T>> buffer(Observable<? extends Opening> bufferOpenings, Func1<Opening, ? extends Observable<? extends Closing>> bufferClosingSelector) {
         return create(OperationBuffer.buffer(this, bufferOpenings, bufferClosingSelector));
     }
 
@@ -1520,7 +1520,7 @@ public Observable<List<T>> buffer(long timespan, long timeshift, TimeUnit unit,
      *         An {@link Observable} which produces connected non-overlapping windows, which are emitted
      *         when the current {@link Observable} created with the {@link Func0} argument produces a {@link rx.util.Closing} object.
      */
-    public Observable<Observable<T>> window(Observable<T> source, Func0<Observable<Closing>> closingSelector) {
+    public Observable<Observable<T>> window(Observable<? extends T> source, Func0<? extends Observable<? extends Closing>> closingSelector) {
         return create(OperationWindow.window(source, closingSelector));
     }
 
@@ -1542,7 +1542,7 @@ public Observable<Observable<T>> window(Observable<T> source, Func0<Observable<C
      * @return
      *         An {@link Observable} which produces windows which are created and emitted when the specified {@link Observable}s publish certain objects.
      */
-    public Observable<Observable<T>> window(Observable<T> source, Observable<Opening> windowOpenings, Func1<Opening, Observable<Closing>> closingSelector) {
+    public Observable<Observable<T>> window(Observable<? extends T> source, Observable<? extends Opening> windowOpenings, Func1<Opening, ? extends Observable<? extends Closing>> closingSelector) {
         return create(OperationWindow.window(source, windowOpenings, closingSelector));
     }
 
@@ -1559,7 +1559,7 @@ public Observable<Observable<T>> window(Observable<T> source, Observable<Opening
      *         An {@link Observable} which produces connected non-overlapping windows containing at most
      *         "count" produced values.
      */
-    public Observable<Observable<T>> window(Observable<T> source, int count) {
+    public Observable<Observable<T>> window(Observable<? extends T> source, int count) {
         return create(OperationWindow.window(source, count));
     }
 
@@ -1579,7 +1579,7 @@ public Observable<Observable<T>> window(Observable<T> source, int count) {
      *         An {@link Observable} which produces windows every "skipped" values containing at most
      *         "count" produced values.
      */
-    public Observable<Observable<T>> window(Observable<T> source, int count, int skip) {
+    public Observable<Observable<T>> window(Observable<? extends T> source, int count, int skip) {
         return create(OperationWindow.window(source, count, skip));
     }
 
@@ -1598,7 +1598,7 @@ public Observable<Observable<T>> window(Observable<T> source, int count, int ski
      * @return
      *         An {@link Observable} which produces connected non-overlapping windows with a fixed duration.
      */
-    public Observable<Observable<T>> window(Observable<T> source, long timespan, TimeUnit unit) {
+    public Observable<Observable<T>> window(Observable<? extends T> source, long timespan, TimeUnit unit) {
         return create(OperationWindow.window(source, timespan, unit));
     }
 
@@ -1619,7 +1619,7 @@ public Observable<Observable<T>> window(Observable<T> source, long timespan, Tim
      * @return
      *         An {@link Observable} which produces connected non-overlapping windows with a fixed duration.
      */
-    public Observable<Observable<T>> window(Observable<T> source, long timespan, TimeUnit unit, Scheduler scheduler) {
+    public Observable<Observable<T>> window(Observable<? extends T> source, long timespan, TimeUnit unit, Scheduler scheduler) {
         return create(OperationWindow.window(source, timespan, unit, scheduler));
     }
 
@@ -1642,7 +1642,7 @@ public Observable<Observable<T>> window(Observable<T> source, long timespan, Tim
      *         An {@link Observable} which produces connected non-overlapping windows which are emitted after
      *         a fixed duration or when the window has reached maximum capacity (which ever occurs first).
      */
-    public Observable<Observable<T>> window(Observable<T> source, long timespan, TimeUnit unit, int count) {
+    public Observable<Observable<T>> window(Observable<? extends T> source, long timespan, TimeUnit unit, int count) {
         return create(OperationWindow.window(source, timespan, unit, count));
     }
 
@@ -1667,7 +1667,7 @@ public Observable<Observable<T>> window(Observable<T> source, long timespan, Tim
      *         An {@link Observable} which produces connected non-overlapping windows which are emitted after
      *         a fixed duration or when the window has reached maximum capacity (which ever occurs first).
      */
-    public Observable<Observable<T>> window(Observable<T> source, long timespan, TimeUnit unit, int count, Scheduler scheduler) {
+    public Observable<Observable<T>> window(Observable<? extends T> source, long timespan, TimeUnit unit, int count, Scheduler scheduler) {
         return create(OperationWindow.window(source, timespan, unit, count, scheduler));
     }
 
@@ -1689,7 +1689,7 @@ public Observable<Observable<T>> window(Observable<T> source, long timespan, Tim
      *         An {@link Observable} which produces new windows periodically, and these are emitted after
      *         a fixed timespan has elapsed.
      */
-    public Observable<Observable<T>> window(Observable<T> source, long timespan, long timeshift, TimeUnit unit) {
+    public Observable<Observable<T>> window(Observable<? extends T> source, long timespan, long timeshift, TimeUnit unit) {
         return create(OperationWindow.window(source, timespan, timeshift, unit));
     }
 
@@ -1713,7 +1713,7 @@ public Observable<Observable<T>> window(Observable<T> source, long timespan, lon
      *         An {@link Observable} which produces new windows periodically, and these are emitted after
      *         a fixed timespan has elapsed.
      */
-    public Observable<Observable<T>> window(Observable<T> source, long timespan, long timeshift, TimeUnit unit, Scheduler scheduler) {
+    public Observable<Observable<T>> window(Observable<? extends T> source, long timespan, long timeshift, TimeUnit unit, Scheduler scheduler) {
         return create(OperationWindow.window(source, timespan, timeshift, unit, scheduler));
     }
 

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

@@ -495,7 +495,7 @@ protected static class ObservableBasedMultiChunkCreator<T, C> implements ChunkCr
 
         private final SafeObservableSubscription subscription = new SafeObservableSubscription();
 
-        public ObservableBasedMultiChunkCreator(final OverlappingChunks<T, C> chunks, Observable<? extends Opening> openings, final Func1<? super Opening, ? extends Observable<? extends Closing>> chunkClosingSelector) {
+        public ObservableBasedMultiChunkCreator(final OverlappingChunks<T, C> chunks, Observable<? extends Opening> openings, final Func1<Opening, ? extends Observable<? extends Closing>> chunkClosingSelector) {
             subscription.wrap(openings.subscribe(new Action1<Opening>() {
                 @Override
                 public void call(Opening opening) {

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

@@ -42,12 +42,14 @@
 
 public final class OperationBuffer extends ChunkedOperation {
 
-    private static final Func0 BUFFER_MAKER = new Func0() {
-        @Override
-        public Object call() {
-            return new Buffer();
-        }
-    };
+    private static <T> Func0<Buffer<T>> bufferMaker() {
+        return new Func0<Buffer<T>>() {
+            @Override
+            public Buffer<T> call() {
+                return new Buffer<T>();
+            }
+        };
+    }
 
     /**
      * <p>This method creates a {@link Func1} object which represents the buffer operation. This operation takes
@@ -74,7 +76,7 @@ public static <T> OnSubscribeFunc<List<T>> buffer(final Observable<T> source, fi
 
             @Override
             public Subscription onSubscribe(Observer<? super List<T>> observer) {
-                NonOverlappingChunks<T, List<T>> buffers = new NonOverlappingChunks<T, List<T>>(observer, BUFFER_MAKER);
+                NonOverlappingChunks<T, List<T>> buffers = new NonOverlappingChunks<T, List<T>>(observer, OperationBuffer.<T>bufferMaker());
                 ChunkCreator creator = new ObservableBasedSingleChunkCreator<T, List<T>>(buffers, bufferClosingSelector);
                 return source.subscribe(new ChunkObserver<T, List<T>>(buffers, observer, creator));
             }
@@ -106,11 +108,11 @@ public Subscription onSubscribe(Observer<? super List<T>> observer) {
      * @return
      *         the {@link Func1} object representing the specified buffer operation.
      */
-    public static <T> OnSubscribeFunc<List<T>> buffer(final Observable<T> source, final Observable<? extends Opening> bufferOpenings, final Func1<? super Opening, ? extends Observable<? extends Closing>> bufferClosingSelector) {
+    public static <T> OnSubscribeFunc<List<T>> buffer(final Observable<T> source, final Observable<? extends Opening> bufferOpenings, final Func1<Opening, ? extends Observable<? extends Closing>> bufferClosingSelector) {
         return new OnSubscribeFunc<List<T>>() {
             @Override
             public Subscription onSubscribe(final Observer<? super List<T>> observer) {
-                OverlappingChunks<T, List<T>> buffers = new OverlappingChunks<T, List<T>>(observer, BUFFER_MAKER);
+                OverlappingChunks<T, List<T>> buffers = new OverlappingChunks<T, List<T>>(observer, OperationBuffer.<T>bufferMaker());
                 ChunkCreator creator = new ObservableBasedMultiChunkCreator<T, List<T>>(buffers, bufferOpenings, bufferClosingSelector);
                 return source.subscribe(new ChunkObserver<T, List<T>>(buffers, observer, creator));
             }
@@ -165,7 +167,7 @@ public static <T> OnSubscribeFunc<List<T>> buffer(final Observable<T> source, fi
         return new OnSubscribeFunc<List<T>>() {
             @Override
             public Subscription onSubscribe(final Observer<? super List<T>> observer) {
-                Chunks<T, List<T>> chunks = new SizeBasedChunks<T, List<T>>(observer, BUFFER_MAKER, count);
+                Chunks<T, List<T>> chunks = new SizeBasedChunks<T, List<T>>(observer, OperationBuffer.<T>bufferMaker(), count);
                 ChunkCreator creator = new SkippingChunkCreator<T, List<T>>(chunks, skip);
                 return source.subscribe(new ChunkObserver<T, List<T>>(chunks, observer, creator));
             }
@@ -220,7 +222,7 @@ public static <T> OnSubscribeFunc<List<T>> buffer(final Observable<T> source, fi
         return new OnSubscribeFunc<List<T>>() {
             @Override
             public Subscription onSubscribe(final Observer<? super List<T>> observer) {
-                NonOverlappingChunks<T, List<T>> buffers = new NonOverlappingChunks<T, List<T>>(observer, BUFFER_MAKER);
+                NonOverlappingChunks<T, List<T>> buffers = new NonOverlappingChunks<T, List<T>>(observer, OperationBuffer.<T>bufferMaker());
                 ChunkCreator creator = new TimeBasedChunkCreator<T, List<T>>(buffers, timespan, unit, scheduler);
                 return source.subscribe(new ChunkObserver<T, List<T>>(buffers, observer, creator));
             }
@@ -281,7 +283,7 @@ public static <T> OnSubscribeFunc<List<T>> buffer(final Observable<T> source, fi
         return new OnSubscribeFunc<List<T>>() {
             @Override
             public Subscription onSubscribe(final Observer<? super List<T>> observer) {
-                Chunks<T, List<T>> chunks = new TimeAndSizeBasedChunks<T, List<T>>(observer, BUFFER_MAKER, count, timespan, unit, scheduler);
+                Chunks<T, List<T>> chunks = new TimeAndSizeBasedChunks<T, List<T>>(observer, OperationBuffer.<T>bufferMaker(), count, timespan, unit, scheduler);
                 ChunkCreator creator = new SingleChunkCreator<T, List<T>>(chunks);
                 return source.subscribe(new ChunkObserver<T, List<T>>(chunks, observer, creator));
             }
@@ -342,7 +344,7 @@ public static <T> OnSubscribeFunc<List<T>> buffer(final Observable<T> source, fi
         return new OnSubscribeFunc<List<T>>() {
             @Override
             public Subscription onSubscribe(final Observer<? super List<T>> observer) {
-                OverlappingChunks<T, List<T>> buffers = new TimeBasedChunks<T, List<T>>(observer, BUFFER_MAKER, timespan, unit, scheduler);
+                OverlappingChunks<T, List<T>> buffers = new TimeBasedChunks<T, List<T>>(observer, OperationBuffer.<T>bufferMaker(), timespan, unit, scheduler);
                 ChunkCreator creator = new TimeBasedChunkCreator<T, List<T>>(buffers, timeshift, unit, scheduler);
                 return source.subscribe(new ChunkObserver<T, List<T>>(buffers, observer, creator));
             }

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

@@ -72,7 +72,7 @@ public Window<T> call() {
      * @return
      *         the {@link rx.util.functions.Func1} object representing the specified window operation.
      */
-    public static <T> OnSubscribeFunc<Observable<T>> window(final Observable<T> source, final Func0<Observable<Closing>> windowClosingSelector) {
+    public static <T> OnSubscribeFunc<Observable<T>> window(final Observable<? extends T> source, final Func0<? extends Observable<? extends Closing>> windowClosingSelector) {
         return new OnSubscribeFunc<Observable<T>>() {
             @Override
             public Subscription onSubscribe(final Observer<? super Observable<T>> observer) {
@@ -109,7 +109,7 @@ public Subscription onSubscribe(final Observer<? super Observable<T>> observer)
      * @return
      *         the {@link rx.util.functions.Func1} object representing the specified window operation.
      */
-    public static <T> OnSubscribeFunc<Observable<T>> window(final Observable<T> source, final Observable<Opening> windowOpenings, final Func1<Opening, Observable<Closing>> windowClosingSelector) {
+    public static <T> OnSubscribeFunc<Observable<T>> window(final Observable<? extends T> source, final Observable<? extends Opening> windowOpenings, final Func1<Opening, ? extends Observable<? extends Closing>> windowClosingSelector) {
         return new OnSubscribeFunc<Observable<T>>() {
             @Override
             public Subscription onSubscribe(final Observer<? super Observable<T>> observer) {
@@ -137,7 +137,7 @@ public Subscription onSubscribe(final Observer<? super Observable<T>> observer)
      * @return
      *         the {@link rx.util.functions.Func1} object representing the specified window operation.
      */
-    public static <T> OnSubscribeFunc<Observable<T>> window(Observable<T> source, int count) {
+    public static <T> OnSubscribeFunc<Observable<T>> window(Observable<? extends T> source, int count) {
         return window(source, count, count);
     }
 
@@ -164,7 +164,7 @@ public static <T> OnSubscribeFunc<Observable<T>> window(Observable<T> source, in
      * @return
      *         the {@link rx.util.functions.Func1} object representing the specified window operation.
      */
-    public static <T> OnSubscribeFunc<Observable<T>> window(final Observable<T> source, final int count, final int skip) {
+    public static <T> OnSubscribeFunc<Observable<T>> window(final Observable<? extends T> source, final int count, final int skip) {
         return new OnSubscribeFunc<Observable<T>>() {
             @Override
             public Subscription onSubscribe(final Observer<? super Observable<T>> observer) {
@@ -194,7 +194,7 @@ public Subscription onSubscribe(final Observer<? super Observable<T>> observer)
      * @return
      *         the {@link rx.util.functions.Func1} object representing the specified window operation.
      */
-    public static <T> OnSubscribeFunc<Observable<T>> window(Observable<T> source, long timespan, TimeUnit unit) {
+    public static <T> OnSubscribeFunc<Observable<T>> window(Observable<? extends T> source, long timespan, TimeUnit unit) {
         return window(source, timespan, unit, Schedulers.threadPoolForComputation());
     }
 
@@ -219,7 +219,7 @@ public static <T> OnSubscribeFunc<Observable<T>> window(Observable<T> source, lo
      * @return
      *         the {@link rx.util.functions.Func1} object representing the specified window operation.
      */
-    public static <T> OnSubscribeFunc<Observable<T>> window(final Observable<T> source, final long timespan, final TimeUnit unit, final Scheduler scheduler) {
+    public static <T> OnSubscribeFunc<Observable<T>> window(final Observable<? extends T> source, final long timespan, final TimeUnit unit, final Scheduler scheduler) {
         return new OnSubscribeFunc<Observable<T>>() {
             @Override
             public Subscription onSubscribe(final Observer<? super Observable<T>> observer) {
@@ -252,7 +252,7 @@ public Subscription onSubscribe(final Observer<? super Observable<T>> observer)
      * @return
      *         the {@link rx.util.functions.Func1} object representing the specified window operation.
      */
-    public static <T> OnSubscribeFunc<Observable<T>> window(Observable<T> source, long timespan, TimeUnit unit, int count) {
+    public static <T> OnSubscribeFunc<Observable<T>> window(Observable<? extends T> source, long timespan, TimeUnit unit, int count) {
         return window(source, timespan, unit, count, Schedulers.threadPoolForComputation());
     }
 
@@ -280,7 +280,7 @@ public static <T> OnSubscribeFunc<Observable<T>> window(Observable<T> source, lo
      * @return
      *         the {@link rx.util.functions.Func1} object representing the specified window operation.
      */
-    public static <T> OnSubscribeFunc<Observable<T>> window(final Observable<T> source, final long timespan, final TimeUnit unit, final int count, final Scheduler scheduler) {
+    public static <T> OnSubscribeFunc<Observable<T>> window(final Observable<? extends T> source, final long timespan, final TimeUnit unit, final int count, final Scheduler scheduler) {
         return new OnSubscribeFunc<Observable<T>>() {
             @Override
             public Subscription onSubscribe(final Observer<? super Observable<T>> observer) {
@@ -313,7 +313,7 @@ public Subscription onSubscribe(final Observer<? super Observable<T>> observer)
      * @return
      *         the {@link rx.util.functions.Func1} object representing the specified window operation.
      */
-    public static <T> OnSubscribeFunc<Observable<T>> window(Observable<T> source, long timespan, long timeshift, TimeUnit unit) {
+    public static <T> OnSubscribeFunc<Observable<T>> window(Observable<? extends T> source, long timespan, long timeshift, TimeUnit unit) {
         return window(source, timespan, timeshift, unit, Schedulers.threadPoolForComputation());
     }
 
@@ -341,7 +341,7 @@ public static <T> OnSubscribeFunc<Observable<T>> window(Observable<T> source, lo
      * @return
      *         the {@link rx.util.functions.Func1} object representing the specified window operation.
      */
-    public static <T> OnSubscribeFunc<Observable<T>> window(final Observable<T> source, final long timespan, final long timeshift, final TimeUnit unit, final Scheduler scheduler) {
+    public static <T> OnSubscribeFunc<Observable<T>> window(final Observable<? extends T> source, final long timespan, final long timeshift, final TimeUnit unit, final Scheduler scheduler) {
         return new OnSubscribeFunc<Observable<T>>() {
             @Override
             public Subscription onSubscribe(final Observer<? super Observable<T>> observer) {