Spring/spring-framework
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Package refactoring in rsocket support

Browse Source 
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.
This commit is contained in:
Rossen Stoyanchev
2019-06-21 06:49:41 +01:00
parent cc05608ae9
commit c6a2c89408
12 changed files with 56 additions and 11 deletions
Download Patch File Download Diff File Expand all files Collapse all files

10
spring-messaging/src/main/java/org/springframework/messaging/handler/invocation/reactive/ChannelSendOperator.java
View File

@@ -189,7 +189,15 @@ class ChannelSendOperator<T> extends Mono<Void> implements Scannable {
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) {

2
spring-messaging/src/main/java/org/springframework/messaging/rsocket/PayloadUtils.java
View File

@@ -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

9
spring-messaging/src/main/java/org/springframework/messaging/rsocket/annotation/package-info.java Normal file
View File

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

21
spring-messaging/src/main/java/org/springframework/messaging/rsocket/MessagingRSocket.java → spring-messaging/src/main/java/org/springframework/messaging/rsocket/annotation/support/MessagingRSocket.java
View File

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

6
spring-messaging/src/main/java/org/springframework/messaging/rsocket/RSocketMessageHandler.java → spring-messaging/src/main/java/org/springframework/messaging/rsocket/annotation/support/RSocketMessageHandler.java
View File

@@ -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;
/**

3
spring-messaging/src/main/java/org/springframework/messaging/rsocket/RSocketPayloadReturnValueHandler.java → spring-messaging/src/main/java/org/springframework/messaging/rsocket/annotation/support/RSocketPayloadReturnValueHandler.java
View File

@@ -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;
/**

3
spring-messaging/src/main/java/org/springframework/messaging/rsocket/RSocketRequesterMethodArgumentResolver.java → spring-messaging/src/main/java/org/springframework/messaging/rsocket/annotation/support/RSocketRequesterMethodArgumentResolver.java
View File

@@ -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;
/**

9
spring-messaging/src/main/java/org/springframework/messaging/rsocket/annotation/support/package-info.java Normal file
View File

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

1
spring-messaging/src/test/java/org/springframework/messaging/rsocket/RSocketBufferLeakTests.java
View File

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

1
spring-messaging/src/test/java/org/springframework/messaging/rsocket/RSocketClientToServerIntegrationTests.java
View File

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

1
spring-messaging/src/test/java/org/springframework/messaging/rsocket/RSocketServerToClientIntegrationTests.java
View File

@@ -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;
/**

1
spring-messaging/src/test/kotlin/org/springframework/messaging/rsocket/RSocketClientToServerCoroutinesIntegrationTests.kt
View File

@@ -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