Package refactoring in rsocket support

rstoyanchev · rstoyanchev · commit c6a2c8940877 · 2019-06-21T06:49:59.000+01:00
Create annotation.support sub-package and move handler code there. This
prepares for a future, functional handler (responder) variant and is
consistent with the package structure under simp.
diff --git a/spring-messaging/src/main/java/org/springframework/messaging/handler/invocation/reactive/ChannelSendOperator.java b/spring-messaging/src/main/java/org/springframework/messaging/handler/invocation/reactive/ChannelSendOperator.java
@@ -189,7 +189,15 @@ public final void onNext(T item) {
 				else if (this.state == State.NEW) {
 					this.item = item;
 					this.state = State.FIRST_SIGNAL_RECEIVED;
-					writeFunction.apply(this).subscribe(this.writeCompletionBarrier);
+					Publisher<Void> result;
+					try {
+						result = writeFunction.apply(this);
+					}
+					catch (Throwable ex) {
+						this.writeCompletionBarrier.onError(ex);
+						return;
+					}
+					result.subscribe(this.writeCompletionBarrier);
 				}
 				else {
 					if (this.subscription != null) {
diff --git a/spring-messaging/src/main/java/org/springframework/messaging/rsocket/PayloadUtils.java b/spring-messaging/src/main/java/org/springframework/messaging/rsocket/PayloadUtils.java
@@ -34,7 +34,7 @@
  * @author Rossen Stoyanchev
  * @since 5.2
  */
-abstract class PayloadUtils {
+public abstract class PayloadUtils {
 
 	/**
 	 * Use this method to slice, retain and wrap the data portion of the
diff --git a/spring-messaging/src/main/java/org/springframework/messaging/rsocket/annotation/package-info.java b/spring-messaging/src/main/java/org/springframework/messaging/rsocket/annotation/package-info.java
@@ -0,0 +1,9 @@
+/**
+ * Annotations and support classes for handling RSocket streams.
+ */
+@NonNullApi
+@NonNullFields
+package org.springframework.messaging.rsocket.annotation;
+
+import org.springframework.lang.NonNullApi;
+import org.springframework.lang.NonNullFields;
diff --git a/spring-messaging/src/main/java/org/springframework/messaging/rsocket/annotation/support/MessagingRSocket.java b/spring-messaging/src/main/java/org/springframework/messaging/rsocket/annotation/support/MessagingRSocket.java
@@ -14,9 +14,11 @@
  * limitations under the License.
  */
 
-package org.springframework.messaging.rsocket;
+package org.springframework.messaging.rsocket.annotation.support;
 
 import java.nio.charset.StandardCharsets;
+import java.util.Arrays;
+import java.util.List;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.function.Function;
 
@@ -39,6 +41,8 @@
 import org.springframework.messaging.MessageHeaders;
 import org.springframework.messaging.handler.DestinationPatternsMessageCondition;
 import org.springframework.messaging.handler.invocation.reactive.HandlerMethodReturnValueHandler;
+import org.springframework.messaging.rsocket.PayloadUtils;
+import org.springframework.messaging.rsocket.RSocketRequester;
 import org.springframework.messaging.support.MessageBuilder;
 import org.springframework.messaging.support.MessageHeaderAccessor;
 import org.springframework.util.Assert;
@@ -57,6 +61,13 @@
  */
 class MessagingRSocket extends AbstractRSocket {
 
+	static final MimeType COMPOSITE_METADATA = new MimeType("message", "x.rsocket.composite-metadata.v0");
+
+	private static final MimeType ROUTING = new MimeType("message", "x.rsocket.routing.v0");
+
+	private static final List<MimeType> METADATA_MIME_TYPES = Arrays.asList(COMPOSITE_METADATA, ROUTING);
+
+
 	private final RSocketMessageHandler messageHandler;
 
 	private final RouteMatcher routeMatcher;
@@ -80,7 +91,7 @@ class MessagingRSocket extends AbstractRSocket {
 		Assert.notNull(dataMimeType, "'dataMimeType' is required");
 		Assert.notNull(metadataMimeType, "'metadataMimeType' is required");
 
-		Assert.isTrue(DefaultRSocketRequester.METADATA_MIME_TYPES.contains(metadataMimeType),
+		Assert.isTrue(METADATA_MIME_TYPES.contains(metadataMimeType),
 				() -> "Unexpected metadatata mime type: '" + metadataMimeType + "'");
 
 		this.messageHandler = messageHandler;
@@ -178,17 +189,17 @@ private Flux<Payload> handleAndReply(Payload firstPayload, Flux<Payload> payload
 	}
 
 	private String getDestination(Payload payload) {
-		if (this.metadataMimeType.equals(DefaultRSocketRequester.COMPOSITE_METADATA)) {
+		if (this.metadataMimeType.equals(COMPOSITE_METADATA)) {
 			CompositeMetadata metadata = new CompositeMetadata(payload.metadata(), false);
 			for (CompositeMetadata.Entry entry : metadata) {
 				String mimeType = entry.getMimeType();
-				if (DefaultRSocketRequester.ROUTING.toString().equals(mimeType)) {
+				if (ROUTING.toString().equals(mimeType)) {
 					return entry.getContent().toString(StandardCharsets.UTF_8);
 				}
 			}
 			return "";
 		}
-		else if (this.metadataMimeType.equals(DefaultRSocketRequester.ROUTING)) {
+		else if (this.metadataMimeType.equals(ROUTING)) {
 			return payload.getMetadataUtf8();
 		}
 		// Should not happen (given constructor assertions)
diff --git a/spring-messaging/src/main/java/org/springframework/messaging/rsocket/annotation/support/RSocketMessageHandler.java b/spring-messaging/src/main/java/org/springframework/messaging/rsocket/annotation/support/RSocketMessageHandler.java
@@ -14,7 +14,7 @@
  * limitations under the License.
  */
 
-package org.springframework.messaging.rsocket;
+package org.springframework.messaging.rsocket.annotation.support;
 
 import java.util.ArrayList;
 import java.util.List;
@@ -33,6 +33,8 @@
 import org.springframework.messaging.MessageDeliveryException;
 import org.springframework.messaging.handler.annotation.reactive.MessageMappingMessageHandler;
 import org.springframework.messaging.handler.invocation.reactive.HandlerMethodReturnValueHandler;
+import org.springframework.messaging.rsocket.RSocketRequester;
+import org.springframework.messaging.rsocket.RSocketStrategies;
 import org.springframework.util.Assert;
 import org.springframework.util.MimeType;
 import org.springframework.util.MimeTypeUtils;
@@ -61,7 +63,7 @@ public class RSocketMessageHandler extends MessageMappingMessageHandler {
 	@Nullable
 	private MimeType defaultDataMimeType;
 
-	private MimeType defaultMetadataMimeType = DefaultRSocketRequester.COMPOSITE_METADATA;
+	private MimeType defaultMetadataMimeType = MessagingRSocket.COMPOSITE_METADATA;
 
 
 	/**
diff --git a/spring-messaging/src/main/java/org/springframework/messaging/rsocket/annotation/support/RSocketPayloadReturnValueHandler.java b/spring-messaging/src/main/java/org/springframework/messaging/rsocket/annotation/support/RSocketPayloadReturnValueHandler.java
@@ -14,7 +14,7 @@
  * limitations under the License.
  */
 
-package org.springframework.messaging.rsocket;
+package org.springframework.messaging.rsocket.annotation.support;
 
 import java.util.List;
 
@@ -30,6 +30,7 @@
 import org.springframework.lang.Nullable;
 import org.springframework.messaging.Message;
 import org.springframework.messaging.handler.invocation.reactive.AbstractEncoderMethodReturnValueHandler;
+import org.springframework.messaging.rsocket.PayloadUtils;
 import org.springframework.util.Assert;
 
 /**
diff --git a/spring-messaging/src/main/java/org/springframework/messaging/rsocket/annotation/support/RSocketRequesterMethodArgumentResolver.java b/spring-messaging/src/main/java/org/springframework/messaging/rsocket/annotation/support/RSocketRequesterMethodArgumentResolver.java
@@ -14,14 +14,15 @@
  * limitations under the License.
  */
 
-package org.springframework.messaging.rsocket;
+package org.springframework.messaging.rsocket.annotation.support;
 
 import io.rsocket.RSocket;
 import reactor.core.publisher.Mono;
 
 import org.springframework.core.MethodParameter;
 import org.springframework.messaging.Message;
 import org.springframework.messaging.handler.invocation.reactive.HandlerMethodArgumentResolver;
+import org.springframework.messaging.rsocket.RSocketRequester;
 import org.springframework.util.Assert;
 
 /**
diff --git a/spring-messaging/src/main/java/org/springframework/messaging/rsocket/annotation/support/package-info.java b/spring-messaging/src/main/java/org/springframework/messaging/rsocket/annotation/support/package-info.java
@@ -0,0 +1,9 @@
+/**
+ * Support classes for working with annotated RSocket stream handling methods.
+ */
+@NonNullApi
+@NonNullFields
+package org.springframework.messaging.rsocket.annotation.support;
+
+import org.springframework.lang.NonNullApi;
+import org.springframework.lang.NonNullFields;
diff --git a/spring-messaging/src/test/java/org/springframework/messaging/rsocket/RSocketBufferLeakTests.java b/spring-messaging/src/test/java/org/springframework/messaging/rsocket/RSocketBufferLeakTests.java
@@ -58,6 +58,7 @@
 import org.springframework.messaging.handler.annotation.MessageExceptionHandler;
 import org.springframework.messaging.handler.annotation.MessageMapping;
 import org.springframework.messaging.handler.annotation.Payload;
+import org.springframework.messaging.rsocket.annotation.support.RSocketMessageHandler;
 import org.springframework.stereotype.Controller;
 import org.springframework.util.ObjectUtils;
 
diff --git a/spring-messaging/src/test/java/org/springframework/messaging/rsocket/RSocketClientToServerIntegrationTests.java b/spring-messaging/src/test/java/org/springframework/messaging/rsocket/RSocketClientToServerIntegrationTests.java
@@ -39,6 +39,7 @@
 import org.springframework.core.io.buffer.NettyDataBufferFactory;
 import org.springframework.messaging.handler.annotation.MessageExceptionHandler;
 import org.springframework.messaging.handler.annotation.MessageMapping;
+import org.springframework.messaging.rsocket.annotation.support.RSocketMessageHandler;
 import org.springframework.stereotype.Controller;
 
 import static org.assertj.core.api.Assertions.assertThat;
diff --git a/spring-messaging/src/test/java/org/springframework/messaging/rsocket/RSocketServerToClientIntegrationTests.java b/spring-messaging/src/test/java/org/springframework/messaging/rsocket/RSocketServerToClientIntegrationTests.java
@@ -44,6 +44,7 @@
 import org.springframework.core.codec.StringDecoder;
 import org.springframework.core.io.buffer.NettyDataBufferFactory;
 import org.springframework.messaging.handler.annotation.MessageMapping;
+import org.springframework.messaging.rsocket.annotation.support.RSocketMessageHandler;
 import org.springframework.stereotype.Controller;
 
 /**
diff --git a/spring-messaging/src/test/kotlin/org/springframework/messaging/rsocket/RSocketClientToServerCoroutinesIntegrationTests.kt b/spring-messaging/src/test/kotlin/org/springframework/messaging/rsocket/RSocketClientToServerCoroutinesIntegrationTests.kt
@@ -37,6 +37,7 @@ import org.springframework.core.codec.StringDecoder
 import org.springframework.core.io.buffer.NettyDataBufferFactory
 import org.springframework.messaging.handler.annotation.MessageExceptionHandler
 import org.springframework.messaging.handler.annotation.MessageMapping
+import org.springframework.messaging.rsocket.annotation.support.RSocketMessageHandler
 import org.springframework.stereotype.Controller
 import reactor.core.publisher.Flux
 import reactor.test.StepVerifier
