diff --git a/spring-integration-http/src/test/java/org/springframework/integration/http/management/IntegrationGraphControllerTests.java b/spring-integration-http/src/test/java/org/springframework/integration/http/management/IntegrationGraphControllerTests.java index fddd7cb5ce..a1e0981685 100644 --- a/spring-integration-http/src/test/java/org/springframework/integration/http/management/IntegrationGraphControllerTests.java +++ b/spring-integration-http/src/test/java/org/springframework/integration/http/management/IntegrationGraphControllerTests.java @@ -82,9 +82,9 @@ public class IntegrationGraphControllerTests { public void testIntegrationGraphGet() throws Exception { this.mockMvc.perform(get("/testIntegration") .header(HttpHeaders.ORIGIN, "https://foo.bar.com") - .accept(MediaType.parseMediaType("application/json;charset=UTF-8"))) + .accept(MediaType.APPLICATION_JSON)) .andExpect(status().isOk()) - .andExpect(content().contentType("application/json;charset=UTF-8")) + .andExpect(content().contentType(MediaType.APPLICATION_JSON_VALUE)) .andExpect(handler().handlerType(IntegrationGraphController.class)) .andExpect(handler().methodName("getGraph")) .andExpect(jsonPath("$.nodes..name") diff --git a/spring-integration-rsocket/src/main/java/org/springframework/integration/rsocket/ClientRSocketConnector.java b/spring-integration-rsocket/src/main/java/org/springframework/integration/rsocket/ClientRSocketConnector.java index b4002d7eb3..1189d877b5 100644 --- a/spring-integration-rsocket/src/main/java/org/springframework/integration/rsocket/ClientRSocketConnector.java +++ b/spring-integration-rsocket/src/main/java/org/springframework/integration/rsocket/ClientRSocketConnector.java @@ -101,11 +101,11 @@ public class ClientRSocketConnector implements InitializingBean, DisposableBean .dataMimeType(this.dataMimeType.toString()); this.factoryConfigurer.accept(clientFactory); clientFactory.setupPayload(this.connectPayload); - this.rsocketMono = clientFactory.transport(this.clientTransport).start(); + this.rsocketMono = clientFactory.transport(this.clientTransport).start().cache(); } public void connect() { - this.rsocketMono.cache().subscribe(); + this.rsocketMono.subscribe(); } public Mono getRSocketRequester() { diff --git a/spring-integration-rsocket/src/main/java/org/springframework/integration/rsocket/inbound/RSocketInboundGateway.java b/spring-integration-rsocket/src/main/java/org/springframework/integration/rsocket/inbound/RSocketInboundGateway.java new file mode 100644 index 0000000000..66b068fdb9 --- /dev/null +++ b/spring-integration-rsocket/src/main/java/org/springframework/integration/rsocket/inbound/RSocketInboundGateway.java @@ -0,0 +1,55 @@ +/* + * Copyright 2019 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.integration.rsocket.inbound; + +import java.util.Arrays; + +import org.springframework.integration.gateway.MessagingGatewaySupport; +import org.springframework.messaging.Message; +import org.springframework.messaging.MessageDeliveryException; +import org.springframework.messaging.ReactiveMessageHandler; +import org.springframework.util.Assert; + +import reactor.core.publisher.Mono; + +/** + * + * @author Artem Bilan + * + * @since 5.2 + */ +public class RSocketInboundGateway extends MessagingGatewaySupport implements ReactiveMessageHandler { + + private final String[] path; + + public RSocketInboundGateway(String... path) { + Assert.notNull(path, "'path' must not be null"); + this.path = path; + } + + + @Override + public Mono handleMessage(Message message) { + if (!isRunning()) { + return Mono.error(new MessageDeliveryException(message, + "The RSocket Inbound Gateway '" + getComponentName() + "' is stopped; " + + "service for path " + Arrays.toString(this.path) + " is not available at the moment.")); + } + return null; + } + +} diff --git a/spring-integration-rsocket/src/main/java/org/springframework/integration/rsocket/outbound/RSocketOutboundGateway.java b/spring-integration-rsocket/src/main/java/org/springframework/integration/rsocket/outbound/RSocketOutboundGateway.java index f31cff994c..7efc5dbd70 100644 --- a/spring-integration-rsocket/src/main/java/org/springframework/integration/rsocket/outbound/RSocketOutboundGateway.java +++ b/spring-integration-rsocket/src/main/java/org/springframework/integration/rsocket/outbound/RSocketOutboundGateway.java @@ -72,7 +72,7 @@ public class RSocketOutboundGateway extends AbstractReplyProducingMessageHandler private Expression commandExpression = new ValueExpression<>(Command.requestResponse); - private Expression publisherElementTypeExpression = new ValueExpression<>(String.class); + private Expression publisherElementTypeExpression; private Expression expectedResponseTypeExpression = new ValueExpression<>(String.class); @@ -181,7 +181,7 @@ public class RSocketOutboundGateway extends AbstractReplyProducingMessageHandler super.doInit(); this.evaluationContext = ExpressionUtils.createStandardEvaluationContext(getBeanFactory()); if (this.clientRSocketConnector != null) { - this.rsocketRequesterMono = this.clientRSocketConnector.getRSocketRequester().cache(); + this.rsocketRequesterMono = this.clientRSocketConnector.getRSocketRequester(); } } @@ -220,8 +220,7 @@ public class RSocketOutboundGateway extends AbstractReplyProducingMessageHandler Message requestMessage) { Object payload = requestMessage.getPayload(); - - if (payload instanceof Publisher) { + if (payload instanceof Publisher && this.publisherElementTypeExpression != null) { Object publisherElementType = evaluateExpressionForType(requestMessage, this.publisherElementTypeExpression, "publisherElementTypeExpression"); return responseSpecForPublisher(requestSpec, (Publisher) payload, publisherElementType); diff --git a/spring-integration-webflux/src/test/java/org/springframework/integration/webflux/inbound/WebFluxInboundEndpointTests.java b/spring-integration-webflux/src/test/java/org/springframework/integration/webflux/inbound/WebFluxInboundEndpointTests.java index ce689f48ab..fd33187708 100644 --- a/spring-integration-webflux/src/test/java/org/springframework/integration/webflux/inbound/WebFluxInboundEndpointTests.java +++ b/spring-integration-webflux/src/test/java/org/springframework/integration/webflux/inbound/WebFluxInboundEndpointTests.java @@ -76,7 +76,7 @@ public class WebFluxInboundEndpointTests { @Test public void testJsonResult() { this.webTestClient.get().uri("/persons") - .accept(MediaType.APPLICATION_JSON_UTF8) + .accept(MediaType.APPLICATION_JSON) .exchange() .expectStatus().isOk() .expectBody() diff --git a/spring-integration-webflux/src/test/java/org/springframework/integration/webflux/management/IntegrationGraphControllerTests.java b/spring-integration-webflux/src/test/java/org/springframework/integration/webflux/management/IntegrationGraphControllerTests.java index b882957e67..e3e0ad20ff 100644 --- a/spring-integration-webflux/src/test/java/org/springframework/integration/webflux/management/IntegrationGraphControllerTests.java +++ b/spring-integration-webflux/src/test/java/org/springframework/integration/webflux/management/IntegrationGraphControllerTests.java @@ -53,10 +53,10 @@ public class IntegrationGraphControllerTests { .build(); webTestClient.get().uri("/testIntegration") - .accept(MediaType.APPLICATION_JSON_UTF8) + .accept(MediaType.APPLICATION_JSON) .exchange() .expectStatus().isOk() - .expectHeader().contentType(MediaType.APPLICATION_JSON_UTF8) + .expectHeader().contentType(MediaType.APPLICATION_JSON) .expectBody() .jsonPath("$.nodes..name").isArray() .jsonPath("$.contentDescriptor.name").isEqualTo("testApplication")