Skip to content

Commit c6a5ea8

Browse files
artembilanartembilan
committed
Polish RSocket module according latest SF
* Add support for the `MetadataExtractor` in the `AbstractRSocketConnector`, `IntegrationRSocketMessageHandler` and `IntegrationRSocket` for more flexible RSocket metadata * Use `DefaultMetadataExtractor` in the `IntegrationRSocketMessageHandler.afterPropertiesSet()` since the getter is not available on the super class yet * Use a `MetadataExtractor` in the `IntegrationRSocket` instead of an old `CompositeMetadata` manipulation * Use `PayloadUtils.retainDataAndReleasePayload()` since it is `public` already * Downgrade to the `rsocketVersion = '0.12.2-RC4'` since it looks like the latest `1.0.0-RC1-SNAPSHOT` doesn't parse a frame properly on the server side
1 parent 5a1846c commit c6a5ea8

File tree

4 files changed

+84
-63
lines changed

4 files changed

+84
-63
lines changed

‎build.gradle

+1-1
Original file line numberDiff line numberDiff line change
@@ -144,7 +144,7 @@ subprojects { subproject ->
144144
reactorVersion = '3.3.0.BUILD-SNAPSHOT'
145145
resilience4jVersion = '0.14.1'
146146
romeToolsVersion = '1.12.0'
147-
rsocketVersion = '1.0.0-RC1-SNAPSHOT'
147+
rsocketVersion = '0.12.2-RC4'
148148
servletApiVersion = '4.0.1'
149149
smackVersion = '4.3.3'
150150
springAmqpVersion = project.hasProperty('springAmqpVersion') ? project.springAmqpVersion : '2.2.0.BUILD-SNAPSHOT'

‎spring-integration-rsocket/src/main/java/org/springframework/integration/rsocket/AbstractRSocketConnector.java

+15
Original file line numberDiff line numberDiff line change
@@ -27,6 +27,8 @@
2727
import org.springframework.core.codec.StringDecoder;
2828
import org.springframework.core.io.buffer.DefaultDataBufferFactory;
2929
import org.springframework.messaging.rsocket.RSocketStrategies;
30+
import org.springframework.messaging.rsocket.annotation.support.DefaultMetadataExtractor;
31+
import org.springframework.messaging.rsocket.annotation.support.MetadataExtractor;
3032
import org.springframework.util.Assert;
3133
import org.springframework.util.MimeType;
3234
import org.springframework.util.MimeTypeUtils;
@@ -121,6 +123,19 @@ public void setEndpoints(IntegrationRSocketEndpoint... endpoints) {
121123
}
122124
}
123125

126+
/**
127+
* Configure a {@link MetadataExtractor} to extract the route and possibly
128+
* other metadata from the first payload of incoming requests.
129+
* <p>By default this is a {@link DefaultMetadataExtractor} with the
130+
* configured {@link RSocketStrategies} (and decoders), extracting a route
131+
* from {@code "message/x.rsocket.routing.v0"} or {@code "text/plain"}
132+
* metadata entries.
133+
* @param extractor the extractor to use
134+
*/
135+
public void setMetadataExtractor(MetadataExtractor extractor) {
136+
this.rSocketMessageHandler.setMetadataExtractor(extractor);
137+
}
138+
124139
/**
125140
* Add an {@link IntegrationRSocketEndpoint} for mapping and handling RSocket requests.
126141
* @param endpoint the {@link IntegrationRSocketEndpoint} to map.

‎spring-integration-rsocket/src/main/java/org/springframework/integration/rsocket/IntegrationRSocket.java

+36-60
Original file line numberDiff line numberDiff line change
@@ -16,9 +16,7 @@
1616

1717
package org.springframework.integration.rsocket;
1818

19-
import java.nio.charset.StandardCharsets;
20-
import java.util.Arrays;
21-
import java.util.List;
19+
import java.util.Map;
2220
import java.util.concurrent.atomic.AtomicBoolean;
2321
import java.util.function.Function;
2422

@@ -28,14 +26,15 @@
2826
import org.springframework.core.io.buffer.DataBufferFactory;
2927
import org.springframework.core.io.buffer.DataBufferUtils;
3028
import org.springframework.core.io.buffer.NettyDataBuffer;
31-
import org.springframework.core.io.buffer.NettyDataBufferFactory;
3229
import org.springframework.lang.Nullable;
3330
import org.springframework.messaging.Message;
3431
import org.springframework.messaging.MessageHeaders;
3532
import org.springframework.messaging.ReactiveMessageHandler;
3633
import org.springframework.messaging.handler.DestinationPatternsMessageCondition;
3734
import org.springframework.messaging.handler.invocation.reactive.HandlerMethodReturnValueHandler;
35+
import org.springframework.messaging.rsocket.PayloadUtils;
3836
import org.springframework.messaging.rsocket.RSocketRequester;
37+
import org.springframework.messaging.rsocket.annotation.support.MetadataExtractor;
3938
import org.springframework.messaging.rsocket.annotation.support.RSocketPayloadReturnValueHandler;
4039
import org.springframework.messaging.rsocket.annotation.support.RSocketRequesterMethodArgumentResolver;
4140
import org.springframework.messaging.support.MessageBuilder;
@@ -44,11 +43,9 @@
4443
import org.springframework.util.MimeType;
4544
import org.springframework.util.RouteMatcher;
4645

47-
import io.netty.buffer.ByteBuf;
4846
import io.rsocket.AbstractRSocket;
4947
import io.rsocket.ConnectionSetupPayload;
5048
import io.rsocket.Payload;
51-
import io.rsocket.metadata.CompositeMetadata;
5249
import reactor.core.publisher.Flux;
5350
import reactor.core.publisher.Mono;
5451
import reactor.core.publisher.MonoProcessor;
@@ -72,11 +69,6 @@ class IntegrationRSocket extends AbstractRSocket {
7269

7370
static final MimeType COMPOSITE_METADATA = new MimeType("message", "x.rsocket.composite-metadata.v0");
7471

75-
static final MimeType ROUTING = new MimeType("message", "x.rsocket.routing.v0");
76-
77-
static final List<MimeType> METADATA_MIME_TYPES = Arrays.asList(COMPOSITE_METADATA, ROUTING);
78-
79-
8072
private final ReactiveMessageHandler handler;
8173

8274
private final RouteMatcher routeMatcher;
@@ -89,24 +81,24 @@ class IntegrationRSocket extends AbstractRSocket {
8981

9082
private final MimeType metadataMimeType;
9183

84+
private final MetadataExtractor metadataExtractor;
85+
9286
IntegrationRSocket(ReactiveMessageHandler handler, RouteMatcher routeMatcher,
9387
RSocketRequester requester, MimeType dataMimeType, MimeType metadataMimeType,
94-
DataBufferFactory bufferFactory) {
88+
MetadataExtractor metadataExtractor, DataBufferFactory bufferFactory) {
9589

9690
Assert.notNull(handler, "'handler' is required");
9791
Assert.notNull(routeMatcher, "'routeMatcher' is required");
9892
Assert.notNull(requester, "'requester' is required");
9993
Assert.notNull(dataMimeType, "'dataMimeType' is required");
10094
Assert.notNull(metadataMimeType, "'metadataMimeType' is required");
10195

102-
Assert.isTrue(METADATA_MIME_TYPES.contains(metadataMimeType),
103-
() -> "Unexpected metadatata mime type: '" + metadataMimeType + "'");
104-
10596
this.handler = handler;
10697
this.routeMatcher = routeMatcher;
10798
this.requester = requester;
10899
this.dataMimeType = dataMimeType;
109100
this.metadataMimeType = metadataMimeType;
101+
this.metadataExtractor = metadataExtractor;
110102
this.bufferFactory = bufferFactory;
111103
}
112104

@@ -163,8 +155,7 @@ public Mono<Void> metadataPush(Payload payload) {
163155

164156

165157
private Mono<Void> handle(Payload payload) {
166-
String destination = getDestination(payload);
167-
MessageHeaders headers = createHeaders(destination, null);
158+
MessageHeaders headers = createHeaders(payload, null);
168159
DataBuffer dataBuffer = retainDataAndReleasePayload(payload);
169160
int refCount = refCount(dataBuffer);
170161
Message<?> message = MessageBuilder.createMessage(dataBuffer, headers);
@@ -176,15 +167,9 @@ private Mono<Void> handle(Payload payload) {
176167
});
177168
}
178169

179-
static int refCount(DataBuffer dataBuffer) {
180-
return dataBuffer instanceof NettyDataBuffer ?
181-
((NettyDataBuffer) dataBuffer).getNativeBuffer().refCnt() : 1;
182-
}
183-
184170
private Flux<Payload> handleAndReply(Payload firstPayload, Flux<Payload> payloads) {
185171
MonoProcessor<Flux<Payload>> replyMono = MonoProcessor.create();
186-
String destination = getDestination(firstPayload);
187-
MessageHeaders headers = createHeaders(destination, replyMono);
172+
MessageHeaders headers = createHeaders(firstPayload, replyMono);
188173

189174
AtomicBoolean read = new AtomicBoolean();
190175
Flux<DataBuffer> buffers =
@@ -206,57 +191,48 @@ private Flux<Payload> handleAndReply(Payload firstPayload, Flux<Payload> payload
206191
}
207192

208193
String getDestination(Payload payload) {
209-
if (this.metadataMimeType.equals(COMPOSITE_METADATA)) {
210-
CompositeMetadata metadata = new CompositeMetadata(payload.metadata(), false);
211-
for (CompositeMetadata.Entry entry : metadata) {
212-
String mimeType = entry.getMimeType();
213-
if (ROUTING.toString().equals(mimeType)) {
214-
return entry.getContent().toString(StandardCharsets.UTF_8);
215-
}
216-
}
217-
return "";
194+
Map<String, Object> metadataValues = this.metadataExtractor.extract(payload, this.metadataMimeType);
195+
Object routingKey = metadataValues.get(MetadataExtractor.ROUTE_KEY);
196+
if (routingKey != null) {
197+
RouteMatcher.Route route = this.routeMatcher.parseRoute(routingKey.toString());
198+
return route.value();
218199
}
219-
else if (this.metadataMimeType.equals(ROUTING)) {
220-
return payload.getMetadataUtf8();
200+
else {
201+
return "";
221202
}
222-
// Should not happen (given constructor assertions)
223-
throw new IllegalArgumentException("Unexpected metadata MimeType");
224203
}
225204

226205
private DataBuffer retainDataAndReleasePayload(Payload payload) {
227-
return payloadToDataBuffer(payload, this.bufferFactory);
206+
payload.retain();
207+
return PayloadUtils.retainDataAndReleasePayload(payload, this.bufferFactory);
228208
}
229209

230-
private MessageHeaders createHeaders(String destination, @Nullable MonoProcessor<?> replyMono) {
210+
private MessageHeaders createHeaders(Payload payload, @Nullable MonoProcessor<?> replyMono) {
231211
MessageHeaderAccessor headers = new MessageHeaderAccessor();
232212
headers.setLeaveMutable(true);
233-
RouteMatcher.Route route = this.routeMatcher.parseRoute(destination);
234-
headers.setHeader(DestinationPatternsMessageCondition.LOOKUP_DESTINATION_HEADER, route);
213+
214+
Map<String, Object> metadataValues = this.metadataExtractor.extract(payload, this.metadataMimeType);
215+
metadataValues.putIfAbsent(MetadataExtractor.ROUTE_KEY, "");
216+
for (Map.Entry<String, Object> entry : metadataValues.entrySet()) {
217+
if (entry.getKey().equals(MetadataExtractor.ROUTE_KEY)) {
218+
RouteMatcher.Route route = this.routeMatcher.parseRoute((String) entry.getValue());
219+
headers.setHeader(DestinationPatternsMessageCondition.LOOKUP_DESTINATION_HEADER, route);
220+
}
221+
else {
222+
headers.setHeader(entry.getKey(), entry.getValue());
223+
}
224+
}
225+
235226
headers.setContentType(this.dataMimeType);
236227
headers.setHeader(RSocketRequesterMethodArgumentResolver.RSOCKET_REQUESTER_HEADER, this.requester);
237-
if (replyMono != null) {
238-
headers.setHeader(RSocketPayloadReturnValueHandler.RESPONSE_HEADER, replyMono);
239-
}
240228
headers.setHeader(HandlerMethodReturnValueHandler.DATA_BUFFER_FACTORY_HEADER, this.bufferFactory);
229+
headers.setHeader(RSocketPayloadReturnValueHandler.RESPONSE_HEADER, replyMono);
230+
241231
return headers.getMessageHeaders();
242232
}
243233

244-
static DataBuffer payloadToDataBuffer(Payload payload, DataBufferFactory bufferFactory) {
245-
payload.retain();
246-
try {
247-
if (bufferFactory instanceof NettyDataBufferFactory) {
248-
ByteBuf byteBuf = payload.sliceData().retain();
249-
return ((NettyDataBufferFactory) bufferFactory).wrap(byteBuf);
250-
}
251-
else {
252-
return bufferFactory.wrap(payload.getData());
253-
}
254-
}
255-
finally {
256-
if (payload.refCnt() > 0) {
257-
payload.release();
258-
}
259-
}
234+
private static int refCount(DataBuffer dataBuffer) {
235+
return dataBuffer instanceof NettyDataBuffer ? ((NettyDataBuffer) dataBuffer).getNativeBuffer().refCnt() : 1;
260236
}
261237

262238
}

‎spring-integration-rsocket/src/main/java/org/springframework/integration/rsocket/IntegrationRSocketMessageHandler.java

+32-2
Original file line numberDiff line numberDiff line change
@@ -32,6 +32,8 @@
3232
import org.springframework.messaging.handler.invocation.reactive.SyncHandlerMethodArgumentResolver;
3333
import org.springframework.messaging.rsocket.RSocketRequester;
3434
import org.springframework.messaging.rsocket.RSocketStrategies;
35+
import org.springframework.messaging.rsocket.annotation.support.DefaultMetadataExtractor;
36+
import org.springframework.messaging.rsocket.annotation.support.MetadataExtractor;
3537
import org.springframework.messaging.rsocket.annotation.support.RSocketMessageHandler;
3638
import org.springframework.util.Assert;
3739
import org.springframework.util.MimeType;
@@ -63,6 +65,8 @@ class IntegrationRSocketMessageHandler extends RSocketMessageHandler {
6365

6466
private MimeType defaultMetadataMimeType = IntegrationRSocket.COMPOSITE_METADATA;
6567

68+
private MetadataExtractor metadataExtractor;
69+
6670
IntegrationRSocketMessageHandler() {
6771
setHandlerPredicate((clazz) -> false);
6872
}
@@ -76,6 +80,7 @@ class IntegrationRSocketMessageHandler extends RSocketMessageHandler {
7680
*/
7781
@Override
7882
public void setDefaultDataMimeType(@Nullable MimeType defaultDataMimeType) {
83+
super.setDefaultDataMimeType(defaultDataMimeType);
7984
this.defaultDataMimeType = defaultDataMimeType;
8085
}
8186

@@ -88,10 +93,25 @@ public void setDefaultDataMimeType(@Nullable MimeType defaultDataMimeType) {
8893
*/
8994
@Override
9095
public void setDefaultMetadataMimeType(MimeType mimeType) {
91-
Assert.notNull(mimeType, "'metadataMimeType' is required");
96+
super.setDefaultMetadataMimeType(mimeType);
9297
this.defaultMetadataMimeType = mimeType;
9398
}
9499

100+
/**
101+
* Configure a {@link MetadataExtractor} to extract the route and possibly
102+
* other metadata from the first payload of incoming requests.
103+
* <p>By default this is a {@link DefaultMetadataExtractor} with the
104+
* configured {@link RSocketStrategies} (and decoders), extracting a route
105+
* from {@code "message/x.rsocket.routing.v0"} or {@code "text/plain"}
106+
* metadata entries.
107+
* @param extractor the extractor to use
108+
*/
109+
@Override
110+
public void setMetadataExtractor(MetadataExtractor extractor) {
111+
super.setMetadataExtractor(extractor);
112+
this.metadataExtractor = extractor;
113+
}
114+
95115
@Override
96116
public BiFunction<ConnectionSetupPayload, RSocket, RSocket> clientAcceptor() {
97117
return this::createRSocket;
@@ -123,6 +143,16 @@ protected List<? extends HandlerMethodArgumentResolver> initArgumentResolvers()
123143
return Collections.singletonList(new MessageHandlerMethodArgumentResolver());
124144
}
125145

146+
@Override
147+
public void afterPropertiesSet() {
148+
super.afterPropertiesSet();
149+
if (this.metadataExtractor == null) {
150+
DefaultMetadataExtractor extractor = new DefaultMetadataExtractor(getRSocketStrategies()); // NOSONAR
151+
extractor.metadataToExtract(MimeTypeUtils.TEXT_PLAIN, String.class, MetadataExtractor.ROUTE_KEY);
152+
this.metadataExtractor = extractor;
153+
}
154+
}
155+
126156
protected IntegrationRSocket createRSocket(ConnectionSetupPayload setupPayload, RSocket rsocket) {
127157
String mimeType = setupPayload.dataMimeType();
128158
MimeType dataMimeType =
@@ -140,7 +170,7 @@ protected IntegrationRSocket createRSocket(ConnectionSetupPayload setupPayload,
140170
Assert.notNull(rSocketStrategies, "No `rSocketStrategies` provided");
141171
RSocketRequester requester = RSocketRequester.wrap(rsocket, dataMimeType, metaMimeType, rSocketStrategies);
142172
return new IntegrationRSocket(this, getRouteMatcher(), requester, dataMimeType, metaMimeType,
143-
rSocketStrategies.dataBufferFactory());
173+
this.metadataExtractor, rSocketStrategies.dataBufferFactory());
144174
}
145175

146176
private static final class MessageHandlerMethodArgumentResolver implements SyncHandlerMethodArgumentResolver {

0 commit comments

Comments
 (0)