More flexible RSocket metadata support

The responding side now relies on a new MetadataExtractor which decodes
metadata entries of interest, and adds them to an output map whose
values are then added as Message headers, and are hence accessible to
controller methods.

Decoded metadata entry values can be added to the output map one for
one, or translated to any number of values (e.g. JSON properties),
as long as one of the resulting pairs has a key called "route".

On the requesting side, now any metadata can be sent, and a String
route for example is not required to be provided explicitly. Instead
an application could create any metadata (e.g. JSON properties) as long
as the server can work out the route from it.

The commit contains further refinements on the requesting side so that
any mime type can be used, not only composite or routing metadata, e.g.
a route in an "text/plain" entry.

Closes gh-23157

spring-messaging/src/test/java/org/springframework/messaging/rsocket/DefaultRSocketRequesterTests.java

@@ -43,11 +43,14 @@ import org.springframework.core.io.buffer.DefaultDataBufferFactory;
 import org.springframework.lang.Nullable;
 import org.springframework.messaging.rsocket.RSocketRequester.RequestSpec;
 import org.springframework.messaging.rsocket.RSocketRequester.ResponseSpec;
-import org.springframework.util.MimeTypeUtils;
 
 import static java.util.concurrent.TimeUnit.MILLISECONDS;
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.assertj.core.api.Assertions.assertThatIllegalArgumentException;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+import static org.springframework.messaging.rsocket.DefaultRSocketRequester.COMPOSITE_METADATA;
+import static org.springframework.messaging.rsocket.DefaultRSocketRequester.ROUTING;
+import static org.springframework.util.MimeTypeUtils.TEXT_PLAIN;
 
 /**
  * Unit tests for {@link DefaultRSocketRequester}.
@@ -75,9 +78,7 @@ public class DefaultRSocketRequesterTests {
 				.encoder(CharSequenceEncoder.allMimeTypes())
 				.build();
 		this.rsocket = new TestRSocket();
-		this.requester = RSocketRequester.wrap(this.rsocket,
-				MimeTypeUtils.TEXT_PLAIN, DefaultRSocketRequester.ROUTING,
-				this.strategies);
+		this.requester = RSocketRequester.wrap(this.rsocket, TEXT_PLAIN, TEXT_PLAIN, this.strategies);
 	}
 
 
@@ -143,13 +144,32 @@ public class DefaultRSocketRequesterTests {
 	}
 
 	@Test
-	public void sendCompositeMetadata() {
-		RSocketRequester requester = RSocketRequester.wrap(this.rsocket,
-				MimeTypeUtils.TEXT_PLAIN, DefaultRSocketRequester.COMPOSITE_METADATA,
-				this.strategies);
+	public void metadataCompositeWithRoute() {
+
+		RSocketRequester requester = RSocketRequester.wrap(
+				this.rsocket, TEXT_PLAIN, COMPOSITE_METADATA, this.strategies);
+
+		requester.route("toA").data("bodyA").send().block(Duration.ofSeconds(5));
+
+		CompositeMetadata entries = new CompositeMetadata(this.rsocket.getSavedPayload().metadata(), false);
+		Iterator<CompositeMetadata.Entry> iterator = entries.iterator();
+
+		assertThat(iterator.hasNext()).isTrue();
+		CompositeMetadata.Entry entry = iterator.next();
+		assertThat(entry.getMimeType()).isEqualTo(ROUTING.toString());
+		assertThat(entry.getContent().toString(StandardCharsets.UTF_8)).isEqualTo("toA");
+
+		assertThat(iterator.hasNext()).isFalse();
+	}
+
+	@Test
+	public void metadataCompositeWithRouteAndTextEntry() {
+
+		RSocketRequester requester = RSocketRequester.wrap(
+				this.rsocket, TEXT_PLAIN, COMPOSITE_METADATA, this.strategies);
 
 		requester.route("toA")
-				.metadata("My metadata", MimeTypeUtils.TEXT_PLAIN).data("bodyA")
+				.metadata("My metadata", TEXT_PLAIN).data("bodyA")
 				.send()
 				.block(Duration.ofSeconds(5));
 
@@ -158,27 +178,46 @@ public class DefaultRSocketRequesterTests {
 
 		assertThat(iterator.hasNext()).isTrue();
 		CompositeMetadata.Entry entry = iterator.next();
-		assertThat(entry.getMimeType()).isEqualTo(DefaultRSocketRequester.ROUTING.toString());
+		assertThat(entry.getMimeType()).isEqualTo(ROUTING.toString());
 		assertThat(entry.getContent().toString(StandardCharsets.UTF_8)).isEqualTo("toA");
 
 		assertThat(iterator.hasNext()).isTrue();
 		entry = iterator.next();
-		assertThat(entry.getMimeType()).isEqualTo(MimeTypeUtils.TEXT_PLAIN.toString());
+		assertThat(entry.getMimeType()).isEqualTo(TEXT_PLAIN.toString());
 		assertThat(entry.getContent().toString(StandardCharsets.UTF_8)).isEqualTo("My metadata");
 
 		assertThat(iterator.hasNext()).isFalse();
 	}
 
+	@Test
+	public void metadataRouteAsText() {
+		RSocketRequester requester = RSocketRequester.wrap(this.rsocket, TEXT_PLAIN, TEXT_PLAIN, this.strategies);
+		requester.route("toA").data("bodyA").send().block(Duration.ofSeconds(5));
+		assertThat(this.rsocket.getSavedPayload().getMetadataUtf8()).isEqualTo("toA");
+	}
+
+	@Test
+	public void metadataAsText() {
+		RSocketRequester requester = RSocketRequester.wrap(this.rsocket, TEXT_PLAIN, TEXT_PLAIN, this.strategies);
+		requester.metadata("toA", null).data("bodyA").send().block(Duration.ofSeconds(5));
+		assertThat(this.rsocket.getSavedPayload().getMetadataUtf8()).isEqualTo("toA");
+	}
+
+	@Test
+	public void metadataMimeTypeMismatch() {
+		RSocketRequester requester = RSocketRequester.wrap(this.rsocket, TEXT_PLAIN, TEXT_PLAIN, this.strategies);
+		assertThatThrownBy(() -> requester.metadata("toA", ROUTING).data("bodyA").send().block())
+				.hasMessageStartingWith("Connection configured for metadata mime type");
+	}
+
 	@Test
 	public void supportedMetadataMimeTypes() {
-		RSocketRequester.wrap(this.rsocket, MimeTypeUtils.TEXT_PLAIN,
-				DefaultRSocketRequester.COMPOSITE_METADATA, this.strategies);
 
-		RSocketRequester.wrap(this.rsocket, MimeTypeUtils.TEXT_PLAIN,
-				DefaultRSocketRequester.ROUTING, this.strategies);
+		RSocketRequester.wrap(this.rsocket, TEXT_PLAIN,
+				COMPOSITE_METADATA, this.strategies);
 
-		assertThatIllegalArgumentException().isThrownBy(() -> RSocketRequester.wrap(
-				this.rsocket, MimeTypeUtils.TEXT_PLAIN, MimeTypeUtils.TEXT_PLAIN, this.strategies));
+		RSocketRequester.wrap(this.rsocket, TEXT_PLAIN,
+				ROUTING, this.strategies);
 	}
 
 	@Test

spring-messaging/src/test/java/org/springframework/messaging/rsocket/LeakAwareNettyDataBufferFactory.java

@@ -0,0 +1,126 @@
+/*
+ * Copyright 2002-2019 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 java.time.Duration;
+import java.time.Instant;
+import java.util.ArrayList;
+import java.util.List;
+
+import io.netty.buffer.ByteBuf;
+import io.netty.buffer.ByteBufAllocator;
+import io.netty.buffer.Unpooled;
+
+import org.springframework.core.io.buffer.DataBuffer;
+import org.springframework.core.io.buffer.NettyDataBuffer;
+import org.springframework.core.io.buffer.NettyDataBufferFactory;
+import org.springframework.core.io.buffer.PooledDataBuffer;
+import org.springframework.util.ObjectUtils;
+
+/**
+ * Unlike {@link org.springframework.core.io.buffer.LeakAwareDataBufferFactory}
+ * this one is an instance of {@link NettyDataBufferFactory} which is necessary
+ * since {@link PayloadUtils} does instanceof checks, and that also allows
+ * intercepting {@link NettyDataBufferFactory#wrap(ByteBuf)}.
+ */
+public class LeakAwareNettyDataBufferFactory extends NettyDataBufferFactory {
+
+	private final List<DataBufferLeakInfo> created = new ArrayList<>();
+
+
+	public LeakAwareNettyDataBufferFactory(ByteBufAllocator byteBufAllocator) {
+		super(byteBufAllocator);
+	}
+
+
+	public void checkForLeaks(Duration duration) throws InterruptedException {
+		Instant start = Instant.now();
+		while (true) {
+			try {
+				this.created.forEach(info -> {
+					if (((PooledDataBuffer) info.getDataBuffer()).isAllocated()) {
+						throw info.getError();
+					}
+				});
+				break;
+			}
+			catch (AssertionError ex) {
+				if (Instant.now().isAfter(start.plus(duration))) {
+					throw ex;
+				}
+			}
+			Thread.sleep(50);
+		}
+	}
+
+	public void reset() {
+		this.created.clear();
+	}
+
+
+	@Override
+	public NettyDataBuffer allocateBuffer() {
+		return (NettyDataBuffer) recordHint(super.allocateBuffer());
+	}
+
+	@Override
+	public NettyDataBuffer allocateBuffer(int initialCapacity) {
+		return (NettyDataBuffer) recordHint(super.allocateBuffer(initialCapacity));
+	}
+
+	@Override
+	public NettyDataBuffer wrap(ByteBuf byteBuf) {
+		NettyDataBuffer dataBuffer = super.wrap(byteBuf);
+		if (byteBuf != Unpooled.EMPTY_BUFFER) {
+			recordHint(dataBuffer);
+		}
+		return dataBuffer;
+	}
+
+	@Override
+	public DataBuffer join(List<? extends DataBuffer> dataBuffers) {
+		return recordHint(super.join(dataBuffers));
+	}
+
+	private DataBuffer recordHint(DataBuffer buffer) {
+		AssertionError error = new AssertionError(String.format(
+				"DataBuffer leak: {%s} {%s} not released.%nStacktrace at buffer creation: ", buffer,
+				ObjectUtils.getIdentityHexString(((NettyDataBuffer) buffer).getNativeBuffer())));
+		this.created.add(new DataBufferLeakInfo(buffer, error));
+		return buffer;
+	}
+
+
+	private static class DataBufferLeakInfo {
+
+		private final DataBuffer dataBuffer;
+
+		private final AssertionError error;
+
+		DataBufferLeakInfo(DataBuffer dataBuffer, AssertionError error) {
+			this.dataBuffer = dataBuffer;
+			this.error = error;
+		}
+
+		DataBuffer getDataBuffer() {
+			return this.dataBuffer;
+		}
+
+		AssertionError getError() {
+			return this.error;
+		}
+	}
+}

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

@@ -18,14 +18,10 @@ package org.springframework.messaging.rsocket;
 
 import java.time.Duration;
 import java.time.Instant;
-import java.util.ArrayList;
 import java.util.List;
 import java.util.concurrent.CopyOnWriteArrayList;
 
-import io.netty.buffer.ByteBuf;
-import io.netty.buffer.ByteBufAllocator;
 import io.netty.buffer.PooledByteBufAllocator;
-import io.netty.buffer.Unpooled;
 import io.netty.util.ReferenceCounted;
 import io.rsocket.AbstractRSocket;
 import io.rsocket.RSocket;
@@ -51,16 +47,11 @@ import org.springframework.context.annotation.Configuration;
 import org.springframework.core.codec.CharSequenceEncoder;
 import org.springframework.core.codec.StringDecoder;
 import org.springframework.core.io.Resource;
-import org.springframework.core.io.buffer.DataBuffer;
-import org.springframework.core.io.buffer.NettyDataBuffer;
-import org.springframework.core.io.buffer.NettyDataBufferFactory;
-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;
 
 import static org.assertj.core.api.Assertions.assertThat;
 
@@ -232,100 +223,6 @@ public class RSocketBufferLeakTests {
 	}
 
 
-	/**
-	 * Unlike {@link org.springframework.core.io.buffer.LeakAwareDataBufferFactory}
-	 * this one is an instance of {@link NettyDataBufferFactory} which is necessary
-	 * since {@link PayloadUtils} does instanceof checks, and that also allows
-	 * intercepting {@link NettyDataBufferFactory#wrap(ByteBuf)}.
-	 */
-	private static class LeakAwareNettyDataBufferFactory extends NettyDataBufferFactory {
-
-		private final List<DataBufferLeakInfo> created = new ArrayList<>();
-
-		LeakAwareNettyDataBufferFactory(ByteBufAllocator byteBufAllocator) {
-			super(byteBufAllocator);
-		}
-
-		void checkForLeaks(Duration duration) throws InterruptedException {
-			Instant start = Instant.now();
-			while (true) {
-				try {
-					this.created.forEach(info -> {
-						if (((PooledDataBuffer) info.getDataBuffer()).isAllocated()) {
-							throw info.getError();
-						}
-					});
-					break;
-				}
-				catch (AssertionError ex) {
-					if (Instant.now().isAfter(start.plus(duration))) {
-						throw ex;
-					}
-				}
-				Thread.sleep(50);
-			}
-		}
-
-		void reset() {
-			this.created.clear();
-		}
-
-
-		@Override
-		public NettyDataBuffer allocateBuffer() {
-			return (NettyDataBuffer) recordHint(super.allocateBuffer());
-		}
-
-		@Override
-		public NettyDataBuffer allocateBuffer(int initialCapacity) {
-			return (NettyDataBuffer) recordHint(super.allocateBuffer(initialCapacity));
-		}
-
-		@Override
-		public NettyDataBuffer wrap(ByteBuf byteBuf) {
-			NettyDataBuffer dataBuffer = super.wrap(byteBuf);
-			if (byteBuf != Unpooled.EMPTY_BUFFER) {
-				recordHint(dataBuffer);
-			}
-			return dataBuffer;
-		}
-
-		@Override
-		public DataBuffer join(List<? extends DataBuffer> dataBuffers) {
-			return recordHint(super.join(dataBuffers));
-		}
-
-		private DataBuffer recordHint(DataBuffer buffer) {
-			AssertionError error = new AssertionError(String.format(
-					"DataBuffer leak: {%s} {%s} not released.%nStacktrace at buffer creation: ", buffer,
-					ObjectUtils.getIdentityHexString(((NettyDataBuffer) buffer).getNativeBuffer())));
-			this.created.add(new DataBufferLeakInfo(buffer, error));
-			return buffer;
-		}
-	}
-
-
-	private static class DataBufferLeakInfo {
-
-		private final DataBuffer dataBuffer;
-
-		private final AssertionError error;
-
-		DataBufferLeakInfo(DataBuffer dataBuffer, AssertionError error) {
-			this.dataBuffer = dataBuffer;
-			this.error = error;
-		}
-
-		DataBuffer getDataBuffer() {
-			return this.dataBuffer;
-		}
-
-		AssertionError getError() {
-			return this.error;
-		}
-	}
-
-
 	/**
 	 * Store all intercepted incoming and outgoing payloads and then use
 	 * {@link #checkForLeaks()} at the end to check reference counts.

spring-messaging/src/test/java/org/springframework/messaging/rsocket/annotation/support/DefaultMetadataExtractorTests.java

@@ -0,0 +1,178 @@
+/*
+ * Copyright 2002-2019 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.annotation.support;
+
+import java.time.Duration;
+import java.util.Map;
+
+import io.netty.buffer.PooledByteBufAllocator;
+import io.rsocket.Payload;
+import io.rsocket.RSocket;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+import org.mockito.ArgumentCaptor;
+import org.mockito.BDDMockito;
+import reactor.core.publisher.Mono;
+
+import org.springframework.core.codec.CharSequenceEncoder;
+import org.springframework.core.codec.StringDecoder;
+import org.springframework.core.io.buffer.DataBufferFactory;
+import org.springframework.messaging.rsocket.LeakAwareNettyDataBufferFactory;
+import org.springframework.messaging.rsocket.RSocketRequester;
+import org.springframework.messaging.rsocket.RSocketStrategies;
+import org.springframework.util.Assert;
+import org.springframework.util.MimeType;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.springframework.messaging.rsocket.annotation.support.MessagingRSocket.COMPOSITE_METADATA;
+import static org.springframework.messaging.rsocket.annotation.support.MessagingRSocket.ROUTING;
+import static org.springframework.messaging.rsocket.annotation.support.MetadataExtractor.ROUTE_KEY;
+import static org.springframework.util.MimeTypeUtils.TEXT_HTML;
+import static org.springframework.util.MimeTypeUtils.TEXT_PLAIN;
+import static org.springframework.util.MimeTypeUtils.TEXT_XML;
+
+
+/**
+ * Unit tests for {@link DefaultMetadataExtractor}.
+ * @author Rossen Stoyanchev
+ */
+public class DefaultMetadataExtractorTests {
+
+	private RSocketStrategies strategies;
+
+	private ArgumentCaptor<Payload> captor;
+
+	private RSocket rsocket;
+
+	private DefaultMetadataExtractor extractor;
+
+
+	@Before
+	public void setUp() {
+		this.strategies = RSocketStrategies.builder()
+				.decoder(StringDecoder.allMimeTypes())
+				.encoder(CharSequenceEncoder.allMimeTypes())
+				.dataBufferFactory(new LeakAwareNettyDataBufferFactory(PooledByteBufAllocator.DEFAULT))
+				.build();
+
+		this.rsocket = BDDMockito.mock(RSocket.class);
+		this.captor = ArgumentCaptor.forClass(Payload.class);
+		BDDMockito.when(this.rsocket.fireAndForget(captor.capture())).thenReturn(Mono.empty());
+
+		this.extractor = new DefaultMetadataExtractor(this.strategies);
+	}
+
+	@After
+	public void tearDown() throws InterruptedException {
+		DataBufferFactory bufferFactory = this.strategies.dataBufferFactory();
+		((LeakAwareNettyDataBufferFactory) bufferFactory).checkForLeaks(Duration.ofSeconds(5));
+	}
+
+
+	@Test
+	public void compositeMetadataWithDefaultSettings() {
+
+		requester(COMPOSITE_METADATA).route("toA")
+				.metadata("text data", TEXT_PLAIN)
+				.metadata("html data", TEXT_HTML)
+				.metadata("xml data", TEXT_XML)
+				.data("data")
+				.send().block();
+
+		Payload payload = this.captor.getValue();
+		Map<String, Object> result = this.extractor.extract(payload, COMPOSITE_METADATA);
+		payload.release();
+
+		assertThat(result).hasSize(1).containsEntry(ROUTE_KEY, "toA");
+	}
+
+	@Test
+	public void compositeMetadataWithMimeTypeRegistrations() {
+
+		this.extractor.metadataToExtract(TEXT_PLAIN, String.class, "text-entry");
+		this.extractor.metadataToExtract(TEXT_HTML, String.class, "html-entry");
+		this.extractor.metadataToExtract(TEXT_XML, String.class, "xml-entry");
+
+		requester(COMPOSITE_METADATA).route("toA")
+				.metadata("text data", TEXT_PLAIN)
+				.metadata("html data", TEXT_HTML)
+				.metadata("xml data", TEXT_XML)
+				.data("data")
+				.send()
+				.block();
+
+		Payload payload = this.captor.getValue();
+		Map<String, Object> result = this.extractor.extract(payload, COMPOSITE_METADATA);
+		payload.release();
+
+		assertThat(result).hasSize(4)
+				.containsEntry(ROUTE_KEY, "toA")
+				.containsEntry("text-entry", "text data")
+				.containsEntry("html-entry", "html data")
+				.containsEntry("xml-entry", "xml data");
+	}
+
+	@Test
+	public void route() {
+
+		requester(ROUTING).route("toA").data("data").send().block();
+		Payload payload = this.captor.getValue();
+		Map<String, Object> result = this.extractor.extract(payload, ROUTING);
+		payload.release();
+
+		assertThat(result).hasSize(1).containsEntry(ROUTE_KEY, "toA");
+	}
+
+	@Test
+	public void routeAsText() {
+
+		this.extractor.metadataToExtract(TEXT_PLAIN, String.class, ROUTE_KEY);
+
+		requester(TEXT_PLAIN).route("toA").data("data").send().block();
+		Payload payload = this.captor.getValue();
+		Map<String, Object> result = this.extractor.extract(payload, TEXT_PLAIN);
+		payload.release();
+
+		assertThat(result).hasSize(1).containsEntry(ROUTE_KEY, "toA");
+	}
+
+	@Test
+	public void routeWithCustomFormatting() {
+
+		this.extractor.metadataToExtract(TEXT_PLAIN, String.class, (text, result) -> {
+			String[] items = text.split(":");
+			Assert.isTrue(items.length == 2, "Expected two items");
+			result.put(ROUTE_KEY, items[0]);
+			result.put("entry1", items[1]);
+		});
+
+		requester(TEXT_PLAIN).metadata("toA:text data", null).data("data").send().block();
+		Payload payload = this.captor.getValue();
+		Map<String, Object> result = this.extractor.extract(payload, TEXT_PLAIN);
+		payload.release();
+
+		assertThat(result).hasSize(2)
+				.containsEntry(ROUTE_KEY, "toA")
+				.containsEntry("entry1", "text data");
+	}
+
+
+	private RSocketRequester requester(MimeType metadataMimeType) {
+		return RSocketRequester.wrap(this.rsocket, TEXT_PLAIN, metadataMimeType, this.strategies);
+	}
+
+}