diff --git a/build.gradle b/build.gradle index eb15e8cadb..19b1b2c92f 100644 --- a/build.gradle +++ b/build.gradle @@ -90,7 +90,7 @@ 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' @@ -98,12 +98,12 @@ ext { 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' diff --git a/spring-integration-core/src/main/java/org/springframework/integration/support/json/MimeTypeSerializer.java b/spring-integration-core/src/main/java/org/springframework/integration/support/json/MimeTypeSerializer.java index 0220dc14ab..c8d7655f80 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/support/json/MimeTypeSerializer.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/support/json/MimeTypeSerializer.java @@ -21,13 +21,12 @@ import java.io.IOException; 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 diff --git a/spring-integration-mongodb/src/main/java/org/springframework/integration/mongodb/dsl/ReactiveMongoDbMessageHandlerSpec.java b/spring-integration-mongodb/src/main/java/org/springframework/integration/mongodb/dsl/ReactiveMongoDbMessageHandlerSpec.java index 2a4a42c850..bebd36c71b 100644 --- a/spring-integration-mongodb/src/main/java/org/springframework/integration/mongodb/dsl/ReactiveMongoDbMessageHandlerSpec.java +++ b/spring-integration-mongodb/src/main/java/org/springframework/integration/mongodb/dsl/ReactiveMongoDbMessageHandlerSpec.java @@ -24,14 +24,13 @@ import org.springframework.data.mongodb.core.convert.MongoConverter; 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 diff --git a/spring-integration-webflux/src/main/java/org/springframework/integration/webflux/outbound/WebFluxRequestExecutingMessageHandler.java b/spring-integration-webflux/src/main/java/org/springframework/integration/webflux/outbound/WebFluxRequestExecutingMessageHandler.java index 2c34c7fa2e..66c1889e33 100644 --- a/spring-integration-webflux/src/main/java/org/springframework/integration/webflux/outbound/WebFluxRequestExecutingMessageHandler.java +++ b/spring-integration-webflux/src/main/java/org/springframework/integration/webflux/outbound/WebFluxRequestExecutingMessageHandler.java @@ -298,38 +298,39 @@ public class WebFluxRequestExecutingMessageHandler extends AbstractHttpRequestEx } private Mono 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 diff --git a/spring-integration-zeromq/src/main/java/org/springframework/integration/zeromq/dsl/ZeroMqMessageHandlerSpec.java b/spring-integration-zeromq/src/main/java/org/springframework/integration/zeromq/dsl/ZeroMqMessageHandlerSpec.java index dbcce5eb0b..0bcc294c39 100644 --- a/spring-integration-zeromq/src/main/java/org/springframework/integration/zeromq/dsl/ZeroMqMessageHandlerSpec.java +++ b/spring-integration-zeromq/src/main/java/org/springframework/integration/zeromq/dsl/ZeroMqMessageHandlerSpec.java @@ -24,7 +24,6 @@ import org.zeromq.ZContext; 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.Message; import org.springframework.messaging.converter.MessageConverter; /** - * The {@link MessageHandlerSpec} extension for {@link ZeroMqMessageHandler}. + * The {@link ReactiveMessageHandlerSpec} extension for {@link ZeroMqMessageHandler}. * * @author Artem Bilan * diff --git a/spring-integration-zeromq/src/main/java/org/springframework/integration/zeromq/dsl/ZeroMqMessageProducerSpec.java b/spring-integration-zeromq/src/main/java/org/springframework/integration/zeromq/dsl/ZeroMqMessageProducerSpec.java index 5b94e3bb8e..9d53fad05f 100644 --- a/spring-integration-zeromq/src/main/java/org/springframework/integration/zeromq/dsl/ZeroMqMessageProducerSpec.java +++ b/spring-integration-zeromq/src/main/java/org/springframework/integration/zeromq/dsl/ZeroMqMessageProducerSpec.java @@ -22,13 +22,10 @@ import java.util.function.Consumer; 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 class ZeroMqMessageProducerSpec } /** - * 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) {