Upgrade to RSocket 1.0 RC7 snapshots

Author: rstoyanchev

build.gradle

@@ -29,7 +29,7 @@ configure(allprojects) { project ->
 			mavenBom "com.fasterxml.jackson:jackson-bom:2.10.3"
 			mavenBom "io.netty:netty-bom:4.1.48.Final"
 			mavenBom "io.projectreactor:reactor-bom:Dysprosium-BUILD-SNAPSHOT"
-			mavenBom "io.rsocket:rsocket-bom:1.0.0-RC6"
+			mavenBom "io.rsocket:rsocket-bom:1.0.0-RC7-SNAPSHOT"
 			mavenBom "org.eclipse.jetty:jetty-bom:9.4.28.v20200408"
 			mavenBom "org.jetbrains.kotlin:kotlin-bom:1.3.72"
 			mavenBom "org.jetbrains.kotlinx:kotlinx-coroutines-bom:1.3.5"
@@ -280,6 +280,7 @@ configure(allprojects) { project ->
 			mavenCentral()
 			maven { url "https://repo.spring.io/libs-spring-framework-build" }
 			maven { url "https://repo.spring.io/snapshot" }
+			maven { url "https://oss.jfrog.org/artifactory/oss-snapshot-local" } // RSocket
 		}
 	}
 	configurations.all {

spring-messaging/src/main/java/org/springframework/messaging/rsocket/ClientRSocketFactoryConfigurer.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2019 the original author or authors.
+ * Copyright 2002-2020 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.
@@ -15,22 +15,24 @@
  */
 package org.springframework.messaging.rsocket;
 
-import io.rsocket.RSocketFactory;
-
 /**
  * Strategy to apply configuration to a client side {@code RSocketFactory}.
  * that's being prepared by {@link RSocketRequester.Builder} to connect
  * to a server.
  *
  * @author Rossen Stoyanchev
  * @since 5.2
+ * @deprecated as of 5.2.6 following the deprecation of
+ * {@link io.rsocket.RSocketFactory.ClientRSocketFactory RSocketFactory.ClientRSocketFactory}
+ * in RSocket 1.0 RC7. Please, use {@link RSocketConnectorConfigurer}.
  */
 @FunctionalInterface
+@Deprecated
 public interface ClientRSocketFactoryConfigurer {
 
 	/**
 	 * Apply configuration to the given {@code ClientRSocketFactory}.
 	 */
-	void configure(RSocketFactory.ClientRSocketFactory rsocketFactory);
+	void configure(io.rsocket.RSocketFactory.ClientRSocketFactory rsocketFactory);
 
 }

spring-messaging/src/main/java/org/springframework/messaging/rsocket/DefaultRSocketRequesterBuilder.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2019 the original author or authors.
+ * Copyright 2002-2020 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.
@@ -25,12 +25,12 @@
 import java.util.function.Consumer;
 
 import io.rsocket.Payload;
-import io.rsocket.RSocketFactory;
 import io.rsocket.frame.decoder.PayloadDecoder;
 import io.rsocket.metadata.WellKnownMimeType;
 import io.rsocket.transport.ClientTransport;
 import io.rsocket.transport.netty.client.TcpClientTransport;
 import io.rsocket.transport.netty.client.WebsocketClientTransport;
+import io.rsocket.util.DefaultPayload;
 import reactor.core.publisher.Mono;
 
 import org.springframework.core.ReactiveAdapter;
@@ -43,6 +43,7 @@
 import org.springframework.core.io.buffer.NettyDataBufferFactory;
 import org.springframework.lang.Nullable;
 import org.springframework.util.Assert;
+import org.springframework.util.ClassUtils;
 import org.springframework.util.CollectionUtils;
 import org.springframework.util.MimeType;
 import org.springframework.util.MimeTypeUtils;
@@ -56,10 +57,17 @@
  */
 final class DefaultRSocketRequesterBuilder implements RSocketRequester.Builder {
 
+	private final static boolean rsocketConnectorPresent =
+			ClassUtils.isPresent("io.rsocket.core.RSocketConnector",
+					DefaultRSocketRequesterBuilder.class.getClassLoader());
+
+
 	private static final Map<String, Object> HINTS = Collections.emptyMap();
 
 	private static final byte[] EMPTY_BYTE_ARRAY = new byte[0];
 
+	private static final Payload EMPTY_SETUP_PAYLOAD = DefaultPayload.create(EMPTY_BYTE_ARRAY);
+
 
 	@Nullable
 	private MimeType dataMimeType;
@@ -84,7 +92,10 @@ final class DefaultRSocketRequesterBuilder implements RSocketRequester.Builder {
 
 	private List<Consumer<RSocketStrategies.Builder>> strategiesConfigurers = new ArrayList<>();
 
-	private List<ClientRSocketFactoryConfigurer> rsocketConfigurers = new ArrayList<>();
+	private List<RSocketConnectorConfigurer> rsocketConnectorConfigurers = new ArrayList<>();
+
+	@SuppressWarnings("deprecation")
+	private List<ClientRSocketFactoryConfigurer> rsocketFactoryConfigurers = new ArrayList<>();
 
 
 	@Override
@@ -133,8 +144,15 @@ public RSocketRequester.Builder rsocketStrategies(Consumer<RSocketStrategies.Bui
 	}
 
 	@Override
+	public RSocketRequester.Builder rsocketConnector(RSocketConnectorConfigurer configurer) {
+		this.rsocketConnectorConfigurers.add(configurer);
+		return this;
+	}
+
+	@Override
+	@Deprecated
 	public RSocketRequester.Builder rsocketFactory(ClientRSocketFactoryConfigurer configurer) {
-		this.rsocketConfigurers.add(configurer);
+		this.rsocketFactoryConfigurers.add(configurer);
 		return this;
 	}
 
@@ -164,28 +182,25 @@ private Mono<RSocketRequester> doConnect(ClientTransport transport) {
 		Assert.isTrue(!rsocketStrategies.encoders().isEmpty(), "No encoders");
 		Assert.isTrue(!rsocketStrategies.decoders().isEmpty(), "No decoders");
 
-		RSocketFactory.ClientRSocketFactory factory = RSocketFactory.connect();
-		this.rsocketConfigurers.forEach(configurer -> configurer.configure(factory));
-
-		if (rsocketStrategies.dataBufferFactory() instanceof NettyDataBufferFactory) {
-			factory.frameDecoder(PayloadDecoder.ZERO_COPY);
-		}
-
 		MimeType metaMimeType = this.metadataMimeType != null ? this.metadataMimeType :
 				MimeTypeUtils.parseMimeType(WellKnownMimeType.MESSAGE_RSOCKET_COMPOSITE_METADATA.getString());
 
 		MimeType dataMimeType = getDataMimeType(rsocketStrategies);
-		factory.dataMimeType(dataMimeType.toString());
-		factory.metadataMimeType(metaMimeType.toString());
-
-		return getSetupPayload(dataMimeType, metaMimeType, rsocketStrategies)
-				.doOnNext(factory::setupPayload)
-				.then(Mono.defer(() ->
-						factory.transport(transport)
-								.start()
-								.map(rsocket -> new DefaultRSocketRequester(
-										rsocket, dataMimeType, metaMimeType, rsocketStrategies))
-				));
+
+		if (rsocketConnectorPresent) {
+			return getSetupPayload(dataMimeType, metaMimeType, rsocketStrategies)
+					.flatMap(payload ->
+							new RSocketConnectorHelper().connect(
+									this.rsocketConnectorConfigurers, this.rsocketFactoryConfigurers,
+									metaMimeType, dataMimeType, payload, rsocketStrategies, transport));
+		}
+		else {
+			return getSetupPayload(dataMimeType, metaMimeType, rsocketStrategies)
+					.flatMap(payload ->
+							new RSocketFactoryHelper().connect(
+									this.rsocketFactoryConfigurers, metaMimeType, dataMimeType, payload,
+									rsocketStrategies, transport));
+		}
 	}
 
 	private RSocketStrategies getRSocketStrategies() {
@@ -234,7 +249,7 @@ private Mono<Payload> getSetupPayload(
 		Object data = this.setupData;
 		boolean hasMetadata = (this.setupRoute != null || !CollectionUtils.isEmpty(this.setupMetadata));
 		if (!hasMetadata && data == null) {
-			return Mono.empty();
+			return Mono.just(EMPTY_SETUP_PAYLOAD);
 		}
 
 		Mono<DataBuffer> dataMono = Mono.empty();
@@ -269,4 +284,69 @@ private Mono<Payload> getSetupPayload(
 				.doOnDiscard(Payload.class, Payload::release);
 	}
 
+
+	private static class RSocketConnectorHelper {
+
+		@SuppressWarnings("deprecation")
+		Mono<RSocketRequester> connect(
+				List<RSocketConnectorConfigurer> connectorConfigurers,
+				List<ClientRSocketFactoryConfigurer> factoryConfigurers,
+				MimeType metaMimeType, MimeType dataMimeType, Payload setupPayload,
+				RSocketStrategies rsocketStrategies, ClientTransport transport) {
+
+			io.rsocket.core.RSocketConnector connector = io.rsocket.core.RSocketConnector.create();
+			connectorConfigurers.forEach(c -> c.configure(connector));
+
+			if (!factoryConfigurers.isEmpty()) {
+				io.rsocket.RSocketFactory.ClientRSocketFactory factory =
+						new io.rsocket.RSocketFactory.ClientRSocketFactory(connector);
+				factoryConfigurers.forEach(c -> c.configure(factory));
+			}
+
+			if (rsocketStrategies.dataBufferFactory() instanceof NettyDataBufferFactory) {
+				connector.payloadDecoder(PayloadDecoder.ZERO_COPY);
+			}
+
+			if (setupPayload != EMPTY_SETUP_PAYLOAD) {
+				connector.setupPayload(setupPayload);
+			}
+
+			return connector
+					.metadataMimeType(metaMimeType.toString())
+					.dataMimeType(dataMimeType.toString())
+					.connect(transport)
+					.map(rsocket -> new DefaultRSocketRequester(
+							rsocket, dataMimeType, metaMimeType, rsocketStrategies));
+		}
+	}
+
+
+	@SuppressWarnings("deprecation")
+	private static class RSocketFactoryHelper {
+
+		Mono<RSocketRequester> connect(
+				List<ClientRSocketFactoryConfigurer> configurers,
+				MimeType metaMimeType, MimeType dataMimeType, Payload setupPayload,
+				RSocketStrategies rsocketStrategies, ClientTransport transport) {
+
+			io.rsocket.RSocketFactory.ClientRSocketFactory factory = io.rsocket.RSocketFactory.connect();
+			configurers.forEach(c -> c.configure(factory));
+
+			if (rsocketStrategies.dataBufferFactory() instanceof NettyDataBufferFactory) {
+				factory.frameDecoder(PayloadDecoder.ZERO_COPY);
+			}
+
+			if (setupPayload != EMPTY_SETUP_PAYLOAD) {
+				factory.setupPayload(setupPayload);
+			}
+
+			return factory.metadataMimeType(metaMimeType.toString())
+					.dataMimeType(dataMimeType.toString())
+					.transport(transport)
+					.start()
+					.map(rsocket -> new DefaultRSocketRequester(
+							rsocket, dataMimeType, metaMimeType, rsocketStrategies));
+		}
+	}
+
 }

spring-messaging/src/main/java/org/springframework/messaging/rsocket/RSocketConnectorConfigurer.java

@@ -0,0 +1,35 @@
+/*
+ * Copyright 2002-2020 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.messaging.rsocket;
+
+import io.rsocket.core.RSocketConnector;
+
+/**
+ * Strategy to apply configuration to an {@code RSocketConnector}. For use with
+ * {@link RSocketRequester.Builder#rsocketConnector RSocketRequester.Builder}.
+ *
+ * @author Rossen Stoyanchev
+ * @since 5.2.6
+ */
+@FunctionalInterface
+public interface RSocketConnectorConfigurer {
+
+	/**
+	 * Apply configuration to the given {@code RSocketConnector}.
+	 */
+	void configure(RSocketConnector connector);
+
+}

spring-messaging/src/main/java/org/springframework/messaging/rsocket/RSocketRequester.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2019 the original author or authors.
+ * Copyright 2002-2020 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.
@@ -185,6 +185,26 @@ interface Builder {
 		 */
 		RSocketRequester.Builder rsocketStrategies(Consumer<RSocketStrategies.Builder> configurer);
 
+		/**
+		 * Callback to configure the {@code RSocketConnector} directly.
+		 * <ul>
+		 * <li>The data and metadata mime types cannot be set directly
+		 * on the {@code RSocketConnector} and will be overridden. Use the
+		 * shortcuts {@link #dataMimeType(MimeType)} and
+		 * {@link #metadataMimeType(MimeType)} on this builder instead.
+		 * <li>The frame decoder also cannot be set directly and instead is set
+		 * to match the configured {@code DataBufferFactory}.
+		 * <li>For the
+		 * {@link io.rsocket.core.RSocketConnector#setupPayload(Payload)
+		 * setupPayload}, consider using methods on this builder to specify the
+		 * route, other metadata, and data as Object values to be encoded.
+		 * <li>To configure client side responding, see
+		 * {@link RSocketMessageHandler#responder(RSocketStrategies, Object...)}.
+		 * </ul>
+		 * @since 5.2.6
+		 */
+		RSocketRequester.Builder rsocketConnector(RSocketConnectorConfigurer configurer);
+
 		/**
 		 * Callback to configure the {@code ClientRSocketFactory} directly.
 		 * <ul>
@@ -201,7 +221,11 @@ interface Builder {
 		 * <li>To configure client side responding, see
 		 * {@link RSocketMessageHandler#clientResponder(RSocketStrategies, Object...)}.
 		 * </ul>
+		 * @deprecated as of 5.2.6 following the deprecation of
+		 * {@link io.rsocket.RSocketFactory.ClientRSocketFactory RSocketFactory.ClientRSocketFactory}
+		 * in RSocket 1.0 RC7. Please, use {@link #rsocketConnector(RSocketConnectorConfigurer)}.
 		 */
+		@Deprecated
 		RSocketRequester.Builder rsocketFactory(ClientRSocketFactoryConfigurer configurer);
 
 		/**

spring-messaging/src/main/java/org/springframework/messaging/rsocket/RSocketStrategies.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2019 the original author or authors.
+ * Copyright 2002-2020 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.
@@ -20,8 +20,6 @@
 import java.util.function.Consumer;
 
 import io.rsocket.Payload;
-import io.rsocket.RSocketFactory.ClientRSocketFactory;
-import io.rsocket.RSocketFactory.ServerRSocketFactory;
 
 import org.springframework.core.ReactiveAdapterRegistry;
 import org.springframework.core.ResolvableType;
@@ -200,9 +198,10 @@ interface Builder {
 		 * <a href="https://github.com/rsocket/rsocket-java#zero-copy">configured</a>
 		 * for zero copy. For client setup, {@link RSocketRequester.Builder}
 		 * adapts automatically to the {@code DataBufferFactory} configured
-		 * here, and sets the frame decoder in {@link ClientRSocketFactory
-		 * ClientRSocketFactory} accordingly. For server setup, the
-		 * {@link ServerRSocketFactory ServerRSocketFactory} must be configured
+		 * here, and sets the frame decoder in
+		 * {@link io.rsocket.core.RSocketConnector RSocketConnector}
+		 * accordingly. For server setup, the
+		 * {@link io.rsocket.core.RSocketServer RSocketServer} must be configured
 		 * accordingly for zero copy too.
 		 * <p>If using {@link DefaultDataBufferFactory} instead, there is no
 		 * need for related config changes in RSocket.