RSocketOutbound refinements; resolve deprecations

* Use properly `cache()` for the `Mono` in the `ClientRSocketConnector`
* Don't require a `publisherElementTypeExpression` in the
`RSocketOutboundGateway`: the target element type is determined
automatically in the `DefaultRequestSpec.toResponseSpec()`
* SF has deprecated `MediaType.APPLICATION_JSON_UTF8`: fix all its usage
in favor of recommendations to use the `MediaType.APPLICATION_JSON` instead
* Also fix plain `application/json;charset=UTF-8` literals to use
`MediaType.APPLICATION_JSON` instead

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")

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<RSocketRequester> getRSocketRequester() {

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<Void> 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;
+	}
+
+}

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);

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()

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")