Improve in generics for some Publisher API

Author: artembilan

spring-integration-core/src/main/java/org/springframework/integration/channel/FluxMessageChannel.java

@@ -45,7 +45,7 @@ public class FluxMessageChannel extends AbstractMessageChannel
 
 	private final List<Subscriber<? super Message<?>>> subscribers = new ArrayList<>();
 
-	private final Map<Publisher<Message<?>>, ConnectableFlux<?>> publishers = new ConcurrentHashMap<>();
+	private final Map<Publisher<? extends Message<?>>, ConnectableFlux<?>> publishers = new ConcurrentHashMap<>();
 
 	private final Flux<Message<?>> flux;
 
@@ -78,7 +78,7 @@ public void subscribe(Subscriber<? super Message<?>> subscriber) {
 	}
 
 	@Override
-	public void subscribeTo(Publisher<Message<?>> publisher) {
+	public void subscribeTo(Publisher<? extends Message<?>> publisher) {
 		ConnectableFlux<?> connectableFlux =
 				Flux.from(publisher)
 						.handle((message, sink) -> sink.next(send(message)))

spring-integration-core/src/main/java/org/springframework/integration/channel/ReactiveStreamsSubscribableChannel.java

@@ -28,6 +28,6 @@
  */
 public interface ReactiveStreamsSubscribableChannel {
 
-	void subscribeTo(Publisher<Message<?>> publisher);
+	void subscribeTo(Publisher<? extends Message<?>> publisher);
 
 }

spring-integration-core/src/main/java/org/springframework/integration/dsl/IntegrationFlowAdapter.java

@@ -228,7 +228,7 @@ protected IntegrationFlowBuilder from(Class<?> serviceInterface,
 		return IntegrationFlows.from(serviceInterface, endpointConfigurer);
 	}
 
-	protected IntegrationFlowBuilder from(Publisher<Message<?>> publisher) {
+	protected IntegrationFlowBuilder from(Publisher<? extends Message<?>> publisher) {
 		return IntegrationFlows.from(publisher);
 	}
 

spring-integration-core/src/main/java/org/springframework/integration/dsl/IntegrationFlows.java

@@ -367,7 +367,7 @@ public static IntegrationFlowBuilder from(Class<?> serviceInterface,
 	 * @param publisher the {@link Publisher} to subscribe to.
 	 * @return new {@link IntegrationFlowBuilder}.
 	 */
-	public static IntegrationFlowBuilder from(Publisher<Message<?>> publisher) {
+	public static IntegrationFlowBuilder from(Publisher<? extends Message<?>> publisher) {
 		FluxMessageChannel reactiveChannel = new FluxMessageChannel();
 		reactiveChannel.subscribeTo(publisher);
 		return from((MessageChannel) reactiveChannel);