Adjust Flux#map mapper to disallow null return values (reactor#4103)

Signed-off-by: Dariusz Jędrzejczyk <dariusz.jedrzejczyk@broadcom.com>

Author: chemicL

reactor-core/src/main/java/reactor/core/publisher/Flux.java

@@ -6623,7 +6623,7 @@ public final Flux<T> log(Logger logger,
 	 *
 	 * @return a transformed {@link Flux}
 	 */
-	public final <V> Flux<V> map(Function<? super T, ? extends @Nullable V> mapper) {
+	public final <V> Flux<V> map(Function<? super T, ? extends V> mapper) {
 		if (this instanceof Fuseable) {
 			return onAssembly(new FluxMapFuseable<>(this, mapper));
 		}

reactor-core/src/main/java/reactor/core/publisher/FluxMap.java

@@ -21,7 +21,6 @@
 
 import org.jspecify.annotations.Nullable;
 import org.reactivestreams.Subscription;
-import reactor.core.CorePublisher;
 import reactor.core.CoreSubscriber;
 import reactor.core.Fuseable;
 
@@ -35,7 +34,7 @@
  */
 final class FluxMap<T, R> extends InternalFluxOperator<T, R> {
 
-	final Function<? super T, ? extends @Nullable R> mapper;
+	final Function<? super T, ? extends R> mapper;
 
 	/**
 	 * Constructs a FluxMap instance with the given source and mapper.
@@ -46,7 +45,7 @@ final class FluxMap<T, R> extends InternalFluxOperator<T, R> {
 	 * @throws NullPointerException if either {@code source} or {@code mapper} is null.
 	 */
 	FluxMap(Flux<? extends T> source,
-			Function<? super T, ? extends @Nullable R> mapper) {
+			Function<? super T, ? extends R> mapper) {
 		super(source);
 		this.mapper = Objects.requireNonNull(mapper, "mapper");
 	}
@@ -72,14 +71,14 @@ static final class MapSubscriber<T, R>
 			implements InnerOperator<T, R> {
 
 		final CoreSubscriber<? super R>        actual;
-		final Function<? super T, ? extends @Nullable R> mapper;
+		final Function<? super T, ? extends R> mapper;
 
 		boolean done;
 
 		Subscription s;
 
 		MapSubscriber(CoreSubscriber<? super R> actual,
-				Function<? super T, ? extends @Nullable R> mapper) {
+				Function<? super T, ? extends R> mapper) {
 			this.actual = actual;
 			this.mapper = mapper;
 		}

reactor-core/src/main/java/reactor/core/publisher/FluxMapFuseable.java

@@ -21,7 +21,6 @@
 
 import org.jspecify.annotations.Nullable;
 import org.reactivestreams.Subscription;
-import reactor.core.CorePublisher;
 import reactor.core.CoreSubscriber;
 import reactor.core.Fuseable;
 
@@ -37,7 +36,7 @@
  */
 final class FluxMapFuseable<T, R> extends InternalFluxOperator<T, R> implements Fuseable {
 
-	final Function<? super T, ? extends @Nullable R> mapper;
+	final Function<? super T, ? extends R> mapper;
 
 	/**
 	 * Constructs a FluxMap instance with the given source and mapper.
@@ -48,7 +47,7 @@ final class FluxMapFuseable<T, R> extends InternalFluxOperator<T, R> implements
 	 * @throws NullPointerException if either {@code source} or {@code mapper} is null.
 	 */
 	FluxMapFuseable(Flux<? extends T> source,
-			Function<? super T, ? extends @Nullable R> mapper) {
+			Function<? super T, ? extends R> mapper) {
 		super(source);
 		this.mapper = Objects.requireNonNull(mapper, "mapper");
 	}
@@ -83,7 +82,7 @@ static final class MapFuseableSubscriber<T, R>
 		int sourceMode;
 
 		MapFuseableSubscriber(CoreSubscriber<? super R> actual,
-				Function<? super T, ? extends @Nullable R> mapper) {
+				Function<? super T, ? extends R> mapper) {
 			this.actual = actual;
 			this.mapper = mapper;
 		}
@@ -242,7 +241,7 @@ static final class MapFuseableConditionalSubscriber<T, R>
 		int sourceMode;
 
 		MapFuseableConditionalSubscriber(ConditionalSubscriber<? super R> actual,
-				Function<? super T, ? extends @Nullable R> mapper) {
+				Function<? super T, ? extends R> mapper) {
 			this.actual = actual;
 			this.mapper = mapper;
 		}