Made the parameter of {Flowable,Observable}.collect(Collector) contravariant on T. (#7033)

Author: cardamon

src/main/java/io/reactivex/rxjava3/core/Flowable.java

@@ -19802,7 +19802,7 @@ public final TestSubscriber<T> test(long initialRequest, boolean cancel) { // No
     @BackpressureSupport(BackpressureKind.UNBOUNDED_IN)
     @SchedulerSupport(SchedulerSupport.NONE)
     @NonNull
-    public final <@NonNull R, A> Single<R> collect(@NonNull Collector<T, A, R> collector) {
+    public final <@NonNull R, A> Single<R> collect(@NonNull Collector<? super T, A, R> collector) {
         Objects.requireNonNull(collector, "collector is null");
         return RxJavaPlugins.onAssembly(new FlowableCollectWithCollectorSingle<>(this, collector));
     }

src/main/java/io/reactivex/rxjava3/core/Observable.java

@@ -16642,7 +16642,7 @@ public final TestObserver<T> test(boolean dispose) { // NoPMD
     @CheckReturnValue
     @SchedulerSupport(SchedulerSupport.NONE)
     @NonNull
-    public final <@NonNull R, A> Single<R> collect(@NonNull Collector<T, A, R> collector) {
+    public final <@NonNull R, A> Single<R> collect(@NonNull Collector<? super T, A, R> collector) {
         Objects.requireNonNull(collector, "collector is null");
         return RxJavaPlugins.onAssembly(new ObservableCollectWithCollectorSingle<>(this, collector));
     }

src/main/java/io/reactivex/rxjava3/internal/jdk8/FlowableCollectWithCollector.java

@@ -36,17 +36,17 @@ public final class FlowableCollectWithCollector<T, A, R> extends Flowable<R> {
 
     final Flowable<T> source;
 
-    final Collector<T, A, R> collector;
+    final Collector<? super T, A, R> collector;
 
-    public FlowableCollectWithCollector(Flowable<T> source, Collector<T, A, R> collector) {
+    public FlowableCollectWithCollector(Flowable<T> source, Collector<? super T, A, R> collector) {
         this.source = source;
         this.collector = collector;
     }
 
     @Override
     protected void subscribeActual(@NonNull Subscriber<? super R> s) {
         A container;
-        BiConsumer<A, T> accumulator;
+        BiConsumer<A, ? super T> accumulator;
         Function<A, R> finisher;
 
         try {

src/main/java/io/reactivex/rxjava3/internal/jdk8/FlowableCollectWithCollectorSingle.java

@@ -39,9 +39,9 @@ public final class FlowableCollectWithCollectorSingle<T, A, R> extends Single<R>
 
     final Flowable<T> source;
 
-    final Collector<T, A, R> collector;
+    final Collector<? super T, A, R> collector;
 
-    public FlowableCollectWithCollectorSingle(Flowable<T> source, Collector<T, A, R> collector) {
+    public FlowableCollectWithCollectorSingle(Flowable<T> source, Collector<? super T, A, R> collector) {
         this.source = source;
         this.collector = collector;
     }
@@ -54,7 +54,7 @@ public Flowable<R> fuseToFlowable() {
     @Override
     protected void subscribeActual(@NonNull SingleObserver<? super R> observer) {
         A container;
-        BiConsumer<A, T> accumulator;
+        BiConsumer<A, ? super T> accumulator;
         Function<A, R> finisher;
 
         try {

src/main/java/io/reactivex/rxjava3/internal/jdk8/ObservableCollectWithCollector.java

@@ -36,17 +36,17 @@ public final class ObservableCollectWithCollector<T, A, R> extends Observable<R>
 
     final Observable<T> source;
 
-    final Collector<T, A, R> collector;
+    final Collector<? super T, A, R> collector;
 
-    public ObservableCollectWithCollector(Observable<T> source, Collector<T, A, R> collector) {
+    public ObservableCollectWithCollector(Observable<T> source, Collector<? super T, A, R> collector) {
         this.source = source;
         this.collector = collector;
     }
 
     @Override
     protected void subscribeActual(@NonNull Observer<? super R> observer) {
         A container;
-        BiConsumer<A, T> accumulator;
+        BiConsumer<A, ? super T> accumulator;
         Function<A, R> finisher;
 
         try {

src/main/java/io/reactivex/rxjava3/internal/jdk8/ObservableCollectWithCollectorSingle.java

@@ -36,9 +36,9 @@ public final class ObservableCollectWithCollectorSingle<T, A, R> extends Single<
 
     final Observable<T> source;
 
-    final Collector<T, A, R> collector;
+    final Collector<? super T, A, R> collector;
 
-    public ObservableCollectWithCollectorSingle(Observable<T> source, Collector<T, A, R> collector) {
+    public ObservableCollectWithCollectorSingle(Observable<T> source, Collector<? super T, A, R> collector) {
         this.source = source;
         this.collector = collector;
     }
@@ -51,7 +51,7 @@ public Observable<R> fuseToObservable() {
     @Override
     protected void subscribeActual(@NonNull SingleObserver<? super R> observer) {
         A container;
-        BiConsumer<A, T> accumulator;
+        BiConsumer<A, ? super T> accumulator;
         Function<A, R> finisher;
 
         try {