diff --git a/spring-messaging/src/main/java/org/springframework/messaging/rsocket/DefaultMetadataExtractor.java b/spring-messaging/src/main/java/org/springframework/messaging/rsocket/DefaultMetadataExtractor.java index a715933a0d..836a7151db 100644 --- a/spring-messaging/src/main/java/org/springframework/messaging/rsocket/DefaultMetadataExtractor.java +++ b/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; /** diff --git a/spring-messaging/src/main/java/org/springframework/messaging/rsocket/DefaultRSocketRequester.java b/spring-messaging/src/main/java/org/springframework/messaging/rsocket/DefaultRSocketRequester.java index 03dce9b453..d708289755 100644 --- a/spring-messaging/src/main/java/org/springframework/messaging/rsocket/DefaultRSocketRequester.java +++ b/spring-messaging/src/main/java/org/springframework/messaging/rsocket/DefaultRSocketRequester.java @@ -215,9 +215,6 @@ final class DefaultRSocketRequester implements RSocketRequester { @SuppressWarnings("unchecked") private 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); diff --git a/spring-messaging/src/main/java/org/springframework/messaging/rsocket/MetadataEncoder.java b/spring-messaging/src/main/java/org/springframework/messaging/rsocket/MetadataEncoder.java index df5f3407d8..cfe849ef70 100644 --- a/spring-messaging/src/main/java/org/springframework/messaging/rsocket/MetadataEncoder.java +++ b/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 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 encoder = this.strategies.encoder(type, mimeType); diff --git a/spring-messaging/src/test/java/org/springframework/messaging/rsocket/MetadataEncoderTests.java b/spring-messaging/src/test/java/org/springframework/messaging/rsocket/MetadataEncoderTests.java index 00c474a5aa..6e951de7e9 100644 --- a/spring-messaging/src/test/java/org/springframework/messaging/rsocket/MetadataEncoderTests.java +++ b/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 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); } }