Various fixes

* Adapt to WebFlux deprecations: spring-projects/spring-framework#25751
* Back to SNAPSHOTs
* Fix new Sonar smells

Author: artembilan

build.gradle

@@ -90,20 +90,20 @@ ext {
 	pahoMqttClientVersion = '1.2.4'
 	postgresVersion = '42.2.14'
 	r2dbch2Version='0.8.4.RELEASE'
-	reactorVersion = '2020.0.0-RC1'
+	reactorVersion = '2020.0.0-SNAPSHOT'
 	resilience4jVersion = '1.5.0'
 	romeToolsVersion = '1.15.0'
 	rsocketVersion = '1.1.0-M2'
 	saajVersion = '1.5.2'
 	servletApiVersion = '4.0.1'
 	smackVersion = '4.3.4'
 	soapVersion = '1.4.0'
-	springAmqpVersion = project.hasProperty('springAmqpVersion') ? project.springAmqpVersion : '2.3.0-M3'
-	springDataVersion = project.hasProperty('springDataVersion') ? project.springDataVersion : '2020.0.0-RC1'
-	springKafkaVersion = '2.6.1'
+	springAmqpVersion = project.hasProperty('springAmqpVersion') ? project.springAmqpVersion : '2.3.0-SNAPSHOT'
+	springDataVersion = project.hasProperty('springDataVersion') ? project.springDataVersion : '2020.0.0-SNAPSHOT'
+	springKafkaVersion = '2.6.2-SNAPSHOT'
 	springRetryVersion = '1.3.0'
 	springSecurityVersion = project.hasProperty('springSecurityVersion') ? project.springSecurityVersion : '5.4.0'
-	springVersion = project.hasProperty('springVersion') ? project.springVersion : '5.3.0-RC1'
+	springVersion = project.hasProperty('springVersion') ? project.springVersion : '5.3.0-SNAPSHOT'
 	springWsVersion = '3.0.10.RELEASE'
 	tomcatVersion = "9.0.38"
 	xstreamVersion = '1.4.13'

spring-integration-core/src/main/java/org/springframework/integration/support/json/MimeTypeSerializer.java

@@ -21,13 +21,12 @@
 import org.springframework.util.MimeType;
 
 import com.fasterxml.jackson.core.JsonGenerator;
-import com.fasterxml.jackson.databind.JsonSerializer;
 import com.fasterxml.jackson.databind.SerializerProvider;
 import com.fasterxml.jackson.databind.jsontype.TypeSerializer;
 import com.fasterxml.jackson.databind.ser.std.StdSerializer;
 
 /**
- * Simple {@link JsonSerializer} extension to represent a {@link MimeType} object in the
+ * Simple {@link StdSerializer} extension to represent a {@link MimeType} object in the
  * target JSON as a plain string.
  *
  * @author Artem Bilan

spring-integration-mongodb/src/main/java/org/springframework/integration/mongodb/dsl/ReactiveMongoDbMessageHandlerSpec.java

@@ -24,14 +24,13 @@
 import org.springframework.expression.Expression;
 import org.springframework.expression.common.LiteralExpression;
 import org.springframework.integration.dsl.ComponentsRegistration;
-import org.springframework.integration.dsl.MessageHandlerSpec;
 import org.springframework.integration.dsl.ReactiveMessageHandlerSpec;
 import org.springframework.integration.expression.FunctionExpression;
 import org.springframework.integration.mongodb.outbound.ReactiveMongoDbStoringMessageHandler;
 import org.springframework.messaging.Message;
 
 /**
- * A {@link MessageHandlerSpec} extension for the Reactive MongoDb Outbound endpoint
+ * A {@link ReactiveMessageHandlerSpec} extension for the Reactive MongoDb Outbound endpoint
  * {@link ReactiveMongoDbStoringMessageHandler}.
  *
  * @author Artem Bilan

spring-integration-webflux/src/main/java/org/springframework/integration/webflux/outbound/WebFluxRequestExecutingMessageHandler.java

@@ -298,38 +298,39 @@ private WebClient.RequestBodySpec createRequestBodySpec(Object uri, HttpMethod h
 	}
 
 	private Mono<ClientResponse> exchangeForResponseMono(WebClient.RequestBodySpec requestSpec) {
-		return requestSpec.exchange()
-				.flatMap(response -> {
+		return requestSpec.retrieve()
+				.onStatus(HttpStatus::isError, response -> {
 					HttpStatus httpStatus = response.statusCode();
-					if (httpStatus.isError()) {
-						return response.body(BodyExtractors.toDataBuffers())
-								.reduce(DataBuffer::write)
-								.map(dataBuffer -> {
-									byte[] bytes = new byte[dataBuffer.readableByteCount()];
-									dataBuffer.read(bytes);
-									DataBufferUtils.release(dataBuffer);
-									return bytes;
-								})
-								.defaultIfEmpty(new byte[0])
-								.map(bodyBytes -> {
-											throw new WebClientResponseException(
-													"ClientResponse has erroneous status code: "
-															+ httpStatus.value() + " "
-															+ httpStatus.getReasonPhrase(),
-													httpStatus.value(),
-													httpStatus.getReasonPhrase(),
-													response.headers().asHttpHeaders(),
-													bodyBytes,
-													response.headers().contentType()
-															.map(MimeType::getCharset)
-															.orElse(StandardCharsets.ISO_8859_1));
-										}
-								);
-					}
-					else {
-						return Mono.just(response);
-					}
-				});
+					return response.body(BodyExtractors.toDataBuffers())
+							.reduce(DataBuffer::write)
+							.map(dataBuffer -> {
+								byte[] bytes = new byte[dataBuffer.readableByteCount()];
+								dataBuffer.read(bytes);
+								DataBufferUtils.release(dataBuffer);
+								return bytes;
+							})
+							.defaultIfEmpty(new byte[0])
+							.map(bodyBytes -> {
+										throw new WebClientResponseException(
+												"ClientResponse has erroneous status code: "
+														+ httpStatus.value() + " "
+														+ httpStatus.getReasonPhrase(),
+												httpStatus.value(),
+												httpStatus.getReasonPhrase(),
+												response.headers().asHttpHeaders(),
+												bodyBytes,
+												response.headers().contentType()
+														.map(MimeType::getCharset)
+														.orElse(StandardCharsets.ISO_8859_1));
+									}
+							);
+				})
+				.toEntityList(DataBuffer.class)
+				.map((entity) ->
+						ClientResponse.create(entity.getStatusCode())
+								.headers((headers) -> headers.addAll(entity.getHeaders()))
+								.body(Flux.fromIterable(entity.getBody())) // NOSONAR - not null according toEntityList()
+								.build());
 	}
 
 	@Nullable

spring-integration-zeromq/src/main/java/org/springframework/integration/zeromq/dsl/ZeroMqMessageHandlerSpec.java

@@ -24,7 +24,6 @@
 import org.zeromq.ZMQ;
 
 import org.springframework.expression.Expression;
-import org.springframework.integration.dsl.MessageHandlerSpec;
 import org.springframework.integration.dsl.ReactiveMessageHandlerSpec;
 import org.springframework.integration.expression.FunctionExpression;
 import org.springframework.integration.mapping.OutboundMessageMapper;
@@ -33,7 +32,7 @@
 import org.springframework.messaging.converter.MessageConverter;
 
 /**
- * The {@link MessageHandlerSpec} extension for {@link ZeroMqMessageHandler}.
+ * The {@link ReactiveMessageHandlerSpec} extension for {@link ZeroMqMessageHandler}.
  *
  * @author Artem Bilan
  *

spring-integration-zeromq/src/main/java/org/springframework/integration/zeromq/dsl/ZeroMqMessageProducerSpec.java

@@ -22,13 +22,10 @@
 import org.zeromq.SocketType;
 import org.zeromq.ZContext;
 import org.zeromq.ZMQ;
-import org.zeromq.ZMsg;
 
 import org.springframework.integration.dsl.MessageProducerSpec;
 import org.springframework.integration.mapping.InboundMessageMapper;
-import org.springframework.integration.zeromq.ZeroMqHeaders;
 import org.springframework.integration.zeromq.inbound.ZeroMqMessageProducer;
-import org.springframework.messaging.Message;
 import org.springframework.messaging.converter.MessageConverter;
 
 /**
@@ -79,9 +76,10 @@ public ZeroMqMessageProducerSpec messageConverter(MessageConverter messageConver
 	}
 
 	/**
-	 * Whether raw {@link ZMsg} is present as a payload of message to produce or
-	 * it is fully converted to a {@link Message} including {@link ZeroMqHeaders#TOPIC} header (if any).
-	 * @param receiveRaw to convert from {@link ZMsg} or not; defaults to convert.
+	 * Whether raw {@link org.zeromq.ZMsg} is present as a payload of message to produce or
+	 * it is fully converted to a {@link org.springframework.messaging.Message} including
+	 * {@link org.springframework.integration.zeromq.ZeroMqHeaders#TOPIC} header (if any).
+	 * @param receiveRaw to convert from {@link org.zeromq.ZMsg} or not; defaults to convert.
 	 * @return the spec
 	 */
 	public ZeroMqMessageProducerSpec receiveRaw(boolean receiveRaw) {