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 index 6206994cf6..297b9ea9a7 100644 --- 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 @@ class ChannelSendOperator extends Mono implements Scannable { else if (this.state == State.NEW) { this.item = item; this.state = State.FIRST_SIGNAL_RECEIVED; - writeFunction.apply(this).subscribe(this.writeCompletionBarrier); + Publisher 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 index e8344aaba4..7c24aa0ce4 100644 --- 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 @@ import org.springframework.core.io.buffer.NettyDataBufferFactory; * @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 new file mode 100644 index 0000000000..ed54c26543 --- /dev/null +++ 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/MessagingRSocket.java b/spring-messaging/src/main/java/org/springframework/messaging/rsocket/annotation/support/MessagingRSocket.java similarity index 91% rename from spring-messaging/src/main/java/org/springframework/messaging/rsocket/MessagingRSocket.java rename to spring-messaging/src/main/java/org/springframework/messaging/rsocket/annotation/support/MessagingRSocket.java index 2ea3fd91eb..70eabdf59d 100644 --- a/spring-messaging/src/main/java/org/springframework/messaging/rsocket/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.Message; 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 @@ import org.springframework.util.RouteMatcher; */ 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 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 @@ class MessagingRSocket extends AbstractRSocket { } 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/RSocketMessageHandler.java b/spring-messaging/src/main/java/org/springframework/messaging/rsocket/annotation/support/RSocketMessageHandler.java similarity index 96% rename from spring-messaging/src/main/java/org/springframework/messaging/rsocket/RSocketMessageHandler.java rename to spring-messaging/src/main/java/org/springframework/messaging/rsocket/annotation/support/RSocketMessageHandler.java index eeff53fc8f..c869c2542f 100644 --- a/spring-messaging/src/main/java/org/springframework/messaging/rsocket/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.Message; 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/RSocketPayloadReturnValueHandler.java b/spring-messaging/src/main/java/org/springframework/messaging/rsocket/annotation/support/RSocketPayloadReturnValueHandler.java similarity index 96% rename from spring-messaging/src/main/java/org/springframework/messaging/rsocket/RSocketPayloadReturnValueHandler.java rename to spring-messaging/src/main/java/org/springframework/messaging/rsocket/annotation/support/RSocketPayloadReturnValueHandler.java index cfdc10fa27..350026b9f8 100644 --- a/spring-messaging/src/main/java/org/springframework/messaging/rsocket/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.core.io.buffer.DataBuffer; 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/RSocketRequesterMethodArgumentResolver.java b/spring-messaging/src/main/java/org/springframework/messaging/rsocket/annotation/support/RSocketRequesterMethodArgumentResolver.java similarity index 94% rename from spring-messaging/src/main/java/org/springframework/messaging/rsocket/RSocketRequesterMethodArgumentResolver.java rename to spring-messaging/src/main/java/org/springframework/messaging/rsocket/annotation/support/RSocketRequesterMethodArgumentResolver.java index 959f155e5d..dbba09d3c2 100644 --- a/spring-messaging/src/main/java/org/springframework/messaging/rsocket/RSocketRequesterMethodArgumentResolver.java +++ b/spring-messaging/src/main/java/org/springframework/messaging/rsocket/annotation/support/RSocketRequesterMethodArgumentResolver.java @@ -14,7 +14,7 @@ * 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; @@ -22,6 +22,7 @@ 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 new file mode 100644 index 0000000000..4c02159a3c --- /dev/null +++ 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 index 18f8290f61..215e0f31a8 100644 --- 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.core.io.buffer.PooledDataBuffer; 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 index 3741492491..4e8ca25db4 100644 --- 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.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 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 index 60b1806711..75cb8666c6 100644 --- 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.CharSequenceEncoder; 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 index 7d66833fa9..c15aa0609f 100644 --- 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