Support ByteBuf as a metadata value

This allows encoding of metadata values externally via some
existing API for encoding RSocket metadata, rather than relying on
registered Encoders.

Also remove explicit checks for DataBuffer since those are supported
through the DataBufferEncoder (registered by default).

See gh-23513

spring-messaging/src/main/java/org/springframework/messaging/rsocket/DefaultMetadataExtractor.java

@@ -37,7 +37,6 @@ import org.springframework.core.codec.Decoder;
 import org.springframework.core.io.buffer.NettyDataBuffer;
 import org.springframework.core.io.buffer.NettyDataBufferFactory;
 import org.springframework.lang.Nullable;
-import org.springframework.util.Assert;
 import org.springframework.util.MimeType;
 
 /**

spring-messaging/src/main/java/org/springframework/messaging/rsocket/DefaultRSocketRequester.java

@@ -215,9 +215,6 @@ final class DefaultRSocketRequester implements RSocketRequester {
 
 		@SuppressWarnings("unchecked")
 		private <T> DataBuffer encodeData(T value, ResolvableType elementType, @Nullable Encoder<?> encoder) {
-			if (value instanceof DataBuffer) {
-				return (DataBuffer) value;
-			}
 			if (encoder == null) {
 				elementType = ResolvableType.forInstance(value);
 				encoder = strategies.encoder(elementType, dataMimeType);

spring-messaging/src/main/java/org/springframework/messaging/rsocket/MetadataEncoder.java

@@ -167,11 +167,14 @@ final class MetadataEncoder {
 					CompositeMetadataFlyweight.encodeAndAddMetadata(composite, this.allocator,
 							WellKnownMimeType.MESSAGE_RSOCKET_ROUTING, encodeRoute());
 				}
-				this.metadata.forEach((value, mimeType) ->
-						CompositeMetadataFlyweight.encodeAndAddMetadata(composite, this.allocator,
-								mimeType.toString(), PayloadUtils.asByteBuf(encodeEntry(value, mimeType))));
+				this.metadata.forEach((value, mimeType) -> {
+					ByteBuf metadata = (value instanceof ByteBuf ?
+							(ByteBuf) value : PayloadUtils.asByteBuf(encodeEntry(value, mimeType)));
+					CompositeMetadataFlyweight.encodeAndAddMetadata(
+							composite, this.allocator, mimeType.toString(), metadata);
+				});
 				return asDataBuffer(composite);
-			}
+				}
 			catch (Throwable ex) {
 				composite.release();
 				throw ex;
@@ -179,7 +182,8 @@ final class MetadataEncoder {
 		}
 		else if (this.route != null) {
 			Assert.isTrue(this.metadata.isEmpty(), "Composite metadata required for route and other entries");
-			return this.metadataMimeType.toString().equals(WellKnownMimeType.MESSAGE_RSOCKET_ROUTING.getString()) ?
+			String routingMimeType = WellKnownMimeType.MESSAGE_RSOCKET_ROUTING.getString();
+			return this.metadataMimeType.toString().equals(routingMimeType) ?
 					asDataBuffer(encodeRoute()) :
 					encodeEntry(this.route, this.metadataMimeType);
 		}
@@ -202,8 +206,8 @@ final class MetadataEncoder {
 
 	@SuppressWarnings("unchecked")
 	private <T> DataBuffer encodeEntry(Object metadata, MimeType mimeType) {
-		if (metadata instanceof DataBuffer) {
-			return (DataBuffer) metadata;
+		if (metadata instanceof ByteBuf) {
+			return asDataBuffer((ByteBuf) metadata);
 		}
 		ResolvableType type = ResolvableType.forInstance(metadata);
 		Encoder<T> encoder = this.strategies.encoder(type, mimeType);

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

@@ -15,13 +15,13 @@
  */
 package org.springframework.messaging.rsocket;
 
-import java.nio.charset.StandardCharsets;
 import java.util.Collections;
 import java.util.Iterator;
 import java.util.Map;
 
 import io.netty.buffer.ByteBuf;
 import io.netty.buffer.ByteBufAllocator;
+import io.netty.buffer.Unpooled;
 import io.rsocket.metadata.CompositeMetadata;
 import io.rsocket.metadata.RoutingMetadata;
 import io.rsocket.metadata.WellKnownMimeType;
@@ -35,6 +35,7 @@ import org.springframework.core.io.buffer.support.DataBufferTestUtils;
 import org.springframework.util.MimeType;
 import org.springframework.util.MimeTypeUtils;
 
+import static java.nio.charset.StandardCharsets.UTF_8;
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.assertj.core.api.Assertions.assertThatThrownBy;
 
@@ -53,27 +54,12 @@ public class MetadataEncoderTests {
 
 
 	@Test
-	public void compositeMetadataWithRoute() {
-		DataBuffer buffer = new MetadataEncoder(COMPOSITE_METADATA, this.strategies)
-				.route("toA")
-				.encode();
-
-		CompositeMetadata entries = new CompositeMetadata(((NettyDataBuffer) buffer).getNativeBuffer(), false);
-		Iterator<CompositeMetadata.Entry> iterator = entries.iterator();
-
-		assertThat(iterator.hasNext()).isTrue();
-		CompositeMetadata.Entry entry = iterator.next();
-		assertThat(entry.getMimeType()).isEqualTo(WellKnownMimeType.MESSAGE_RSOCKET_ROUTING.getString());
-		assertRoute("toA", entry.getContent());
-		assertThat(iterator.hasNext()).isFalse();
-	}
-
-	@Test
-	public void compositeMetadataWithRouteAndText() {
+	public void compositeMetadata() {
 
 		DataBuffer buffer = new MetadataEncoder(COMPOSITE_METADATA, this.strategies)
 				.route("toA")
 				.metadata("My metadata", MimeTypeUtils.TEXT_PLAIN)
+				.metadata(Unpooled.wrappedBuffer("Raw data".getBytes(UTF_8)), MimeTypeUtils.APPLICATION_OCTET_STREAM)
 				.encode();
 
 		CompositeMetadata entries = new CompositeMetadata(((NettyDataBuffer) buffer).getNativeBuffer(), false);
@@ -86,19 +72,25 @@ public class MetadataEncoderTests {
 
 		assertThat(iterator.hasNext()).isTrue();
 		entry = iterator.next();
-		assertThat(entry.getMimeType()).isEqualTo(MimeTypeUtils.TEXT_PLAIN.toString());
-		assertThat(entry.getContent().toString(StandardCharsets.UTF_8)).isEqualTo("My metadata");
+		assertThat(entry.getMimeType()).isEqualTo(MimeTypeUtils.TEXT_PLAIN_VALUE);
+		assertThat(entry.getContent().toString(UTF_8)).isEqualTo("My metadata");
+
+		assertThat(iterator.hasNext()).isTrue();
+		entry = iterator.next();
+		assertThat(entry.getMimeType()).isEqualTo(MimeTypeUtils.APPLICATION_OCTET_STREAM_VALUE);
+		assertThat(entry.getContent().toString(UTF_8)).isEqualTo("Raw data");
 
 		assertThat(iterator.hasNext()).isFalse();
 	}
 
 	@Test
 	public void routeWithRoutingMimeType() {
-		MimeType metaMimeType = MimeTypeUtils.parseMimeType(
+
+		MimeType mimeType = MimeTypeUtils.parseMimeType(
 				WellKnownMimeType.MESSAGE_RSOCKET_ROUTING.getString());
 
 		DataBuffer buffer =
-				new MetadataEncoder(metaMimeType, this.strategies)
+				new MetadataEncoder(mimeType, this.strategies)
 						.route("toA")
 						.encode();
 
@@ -127,6 +119,16 @@ public class MetadataEncoderTests {
 
 	@Test
 	public void metadataWithTextPlainMimeType() {
+		DataBuffer buffer =
+				new MetadataEncoder(MimeTypeUtils.TEXT_PLAIN, this.strategies)
+						.metadata(Unpooled.wrappedBuffer("Raw data".getBytes(UTF_8)), null)
+						.encode();
+
+		assertThat(dumpString(buffer)).isEqualTo("Raw data");
+	}
+
+	@Test
+	public void metadataWithByteBuf() {
 		DataBuffer buffer =
 				new MetadataEncoder(MimeTypeUtils.TEXT_PLAIN, this.strategies)
 						.metadata("toA", null)
@@ -210,7 +212,7 @@ public class MetadataEncoderTests {
 	}
 
 	private String dumpString(DataBuffer buffer) {
-		return DataBufferTestUtils.dumpString(buffer, StandardCharsets.UTF_8);
+		return DataBufferTestUtils.dumpString(buffer, UTF_8);
 	}
 
 }