Ensure read/write indices of input buffers are not modified

Author: rdegnan

rsocket-core/src/main/java/io/rsocket/frame/FrameHeaderFlyweight.java

@@ -115,7 +115,7 @@ public static int encodeMetadata(
         encodeLength(byteBuf, metadataOffset, metadataLength);
         length += FRAME_LENGTH_SIZE;
       }
-      byteBuf.setBytes(metadataOffset + length, metadata);
+      byteBuf.setBytes(metadataOffset + length, metadata, metadata.readerIndex(), metadataLength);
       length += metadataLength;
     }
 
@@ -127,7 +127,7 @@ public static int encodeData(final ByteBuf byteBuf, final int dataOffset, final
     final int dataLength = data.readableBytes();
 
     if (0 < dataLength) {
-      byteBuf.setBytes(dataOffset, data);
+      byteBuf.setBytes(dataOffset, data, data.readerIndex(), dataLength);
       length += dataLength;
     }
 

rsocket-core/src/main/java/io/rsocket/frame/SetupFrameFlyweight.java

@@ -134,7 +134,7 @@ static int encode(
       byteBuf.setShort(length, resumeToken.readableBytes());
       length += Short.BYTES;
       int resumeTokenLength = resumeToken.readableBytes();
-      byteBuf.setBytes(length, resumeToken, resumeTokenLength);
+      byteBuf.setBytes(length, resumeToken, resumeToken.readerIndex(), resumeTokenLength);
       length += resumeTokenLength;
     }
 

rsocket-core/src/main/java/io/rsocket/util/ByteBufPayload.java

@@ -51,12 +51,12 @@ public boolean hasMetadata() {
 
   @Override
   public ByteBuf sliceMetadata() {
-    return metadata == null ? Unpooled.EMPTY_BUFFER : metadata;
+    return metadata == null ? Unpooled.EMPTY_BUFFER : metadata.slice();
   }
 
   @Override
   public ByteBuf sliceData() {
-    return data;
+    return data.slice();
   }
 
   @Override

rsocket-core/src/main/java/io/rsocket/util/DefaultPayload.java

@@ -57,12 +57,12 @@ public ByteBuf sliceData() {
 
   @Override
   public ByteBuffer getMetadata() {
-    return metadata == null ? DefaultPayload.EMPTY_BUFFER : metadata;
+    return metadata == null ? DefaultPayload.EMPTY_BUFFER : metadata.duplicate();
   }
 
   @Override
   public ByteBuffer getData() {
-    return data;
+    return data.duplicate();
   }
 
   @Override