4.x: Cleanup, fix inferable types via var or diamond (#8189)

Author: akarnokd

src/main/java/io/reactivex/rxjava4/core/Streamable.java

@@ -115,7 +115,7 @@ static <T> Streamable<T> fromPublisher(@NonNull Flow.Publisher<T> source) {
     @NonNull
     static <T> Streamable<T> fromPublisher(@NonNull Flow.Publisher<T> source, @NonNull ExecutorService executor) {
         Objects.requireNonNull(source, "source is null");
-        return new StreamableFromPublisher<T>(source, executor);
+        return new StreamableFromPublisher<>(source, executor);
     }
 
     /**
@@ -357,7 +357,7 @@ default CompletionStageDisposable<Void> forEach(@NonNull Consumer<? super T> con
             return null;
         });
         canceller.add(Disposable.fromFuture(future));
-        return new CompletionStageDisposable<Void>(StreamableHelper.toCompletionStage((Future<Void>)(Future<?>)future), canceller);
+        return new CompletionStageDisposable<>(StreamableHelper.toCompletionStage((Future<Void>) (Future<?>) future), canceller);
     }
 
     /**
@@ -407,8 +407,8 @@ default CompletionStageDisposable<Void> forEach(
             return null;
         });
         canceller.add(Disposable.fromFuture(future));
-        return new CompletionStageDisposable<Void>(
-                StreamableHelper.toCompletionStage((Future<Void>)(Future<?>)future), canceller);
+        return new CompletionStageDisposable<>(
+                StreamableHelper.toCompletionStage((Future<Void>) (Future<?>) future), canceller);
     }
 
     /**

src/main/java/io/reactivex/rxjava4/internal/operators/flowable/FlowableCache.java

@@ -103,7 +103,7 @@ public FlowableCache(Flowable<T> source, int capacityHint) {
         Node<T> n = new Node<>(capacityHint);
         this.head = n;
         this.tail = n;
-        this.subscribers = new AtomicReference<>(EMPTY);
+        this.subscribers = new AtomicReference<CacheSubscription<T>[]>(EMPTY);
     }
 
     @Override

src/main/java/io/reactivex/rxjava4/internal/operators/flowable/FlowablePublish.java

@@ -179,7 +179,7 @@ static final class PublishConnection<T>
             this.upstream = new AtomicReference<>();
             this.connect = new AtomicBoolean();
             this.bufferSize = bufferSize;
-            this.subscribers = new AtomicReference<>(EMPTY);
+            this.subscribers = new AtomicReference<InnerSubscription<T>[]>(EMPTY);
         }
 
         @SuppressWarnings("unchecked")

src/main/java/io/reactivex/rxjava4/internal/operators/flowable/FlowablePublishMulticast.java

@@ -158,7 +158,7 @@ static final class MulticastProcessor<T> extends Flowable<T> implements Flowable
             this.delayError = delayError;
             this.wip = new AtomicInteger();
             this.upstream = new AtomicReference<>();
-            this.subscribers = new AtomicReference<>(EMPTY);
+            this.subscribers = new AtomicReference<MulticastSubscription<T>[]>(EMPTY);
         }
 
         @Override

src/main/java/io/reactivex/rxjava4/internal/operators/maybe/MaybeCache.java

@@ -43,7 +43,7 @@ public final class MaybeCache<T> extends Maybe<T> implements MaybeObserver<T> {
     @SuppressWarnings("unchecked")
     public MaybeCache(MaybeSource<T> source) {
         this.source = new AtomicReference<>(source);
-        this.observers = new AtomicReference<>(EMPTY);
+        this.observers = new AtomicReference<CacheDisposable<T>[]>(EMPTY);
     }
 
     @Override

src/main/java/io/reactivex/rxjava4/internal/operators/single/SingleCache.java

@@ -40,7 +40,7 @@ public final class SingleCache<T> extends Single<T> implements SingleObserver<T>
     public SingleCache(SingleSource<? extends T> source) {
         this.source = source;
         this.wip = new AtomicInteger();
-        this.observers = new AtomicReference<>(EMPTY);
+        this.observers = new AtomicReference<CacheDisposable<T>[]>(EMPTY);
     }
 
     @Override

src/main/java/io/reactivex/rxjava4/internal/operators/single/SingleOnErrorComplete.java

@@ -38,6 +38,6 @@ public SingleOnErrorComplete(Single<T> source,
 
     @Override
     protected void subscribeActual(MaybeObserver<? super T> observer) {
-        source.subscribe(new MaybeOnErrorComplete.OnErrorCompleteMultiObserver<T>(observer, predicate));
+        source.subscribe(new MaybeOnErrorComplete.OnErrorCompleteMultiObserver<>(observer, predicate));
     }
 }

src/main/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableEmpty.java

@@ -24,7 +24,7 @@ public final class StreamableEmpty<T> implements Streamable<T> {
 
     @Override
     public @NonNull Streamer<@NonNull T> stream(@NonNull DisposableContainer cancellation) {
-        return new EmptyStreamer<T>();
+        return new EmptyStreamer<>();
     }
 
     static final class EmptyStreamer<T> implements Streamer<T> {

src/main/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableJust.java

@@ -28,7 +28,7 @@ public StreamableJust(T item) {
 
     @Override
     public @NonNull Streamer<@NonNull T> stream(@NonNull DisposableContainer cancellation) {
-        return new JustStreamer<T>(item, cancellation);
+        return new JustStreamer<>(item, cancellation);
     }
 
     static final class JustStreamer<T> implements Streamer<T>, Disposable {

src/main/java/io/reactivex/rxjava4/processors/AsyncProcessor.java

@@ -148,7 +148,7 @@ public static <T> AsyncProcessor<T> create() {
      */
     @SuppressWarnings("unchecked")
     AsyncProcessor() {
-        this.subscribers = new AtomicReference<>(EMPTY);
+        this.subscribers = new AtomicReference<AsyncSubscription<T>[]>(EMPTY);
     }
 
     @Override