NettyReactiveStreamsBody shouldn't assume ByteBuffer has an array without offset and partial length

Author: slandelle

client/src/main/java/org/asynchttpclient/netty/request/body/NettyReactiveStreamsBody.java

@@ -81,7 +81,7 @@ public void onSubscribe(Subscription s) {
         }
         @Override
         public void onNext(ByteBuffer t) {
-            ByteBuf buffer = Unpooled.wrappedBuffer(t.array());
+            ByteBuf buffer = Unpooled.wrappedBuffer(t);
             HttpContent content = new DefaultHttpContent(buffer);
             subscriber.onNext(content);
         }

client/src/test/java/org/asynchttpclient/reactivestreams/ReactiveStreamsTest.java

@@ -12,16 +12,15 @@
  */
 package org.asynchttpclient.reactivestreams;
 
-import static org.asynchttpclient.Dsl.*;
-import static org.asynchttpclient.test.TestUtils.*;
 import static io.netty.handler.codec.http.HttpHeaderNames.*;
+import static org.asynchttpclient.Dsl.*;
+import static org.asynchttpclient.test.TestUtils.LARGE_IMAGE_BYTES;
 import static org.testng.Assert.assertEquals;
 import io.netty.handler.codec.http.HttpHeaders;
 
 import java.io.ByteArrayOutputStream;
 import java.nio.ByteBuffer;
 import java.util.ArrayList;
-import java.util.Arrays;
 import java.util.Collections;
 import java.util.Iterator;
 import java.util.List;
@@ -359,10 +358,10 @@ public boolean hasNext() {
 
                 @Override
                 public ByteBuffer next() {
-                    int newIndex = Math.min(currentIndex + chunkSize, payload.length);
-                    byte[] bytesInElement = Arrays.copyOfRange(payload, currentIndex, newIndex);
-                    currentIndex = newIndex;
-                    return ByteBuffer.wrap(bytesInElement);
+                    int thisCurrentIndex = currentIndex;
+                    int length = Math.min(chunkSize, payload.length - thisCurrentIndex);
+                    currentIndex += length;
+                    return ByteBuffer.wrap(payload, thisCurrentIndex, length);
                 }
 
                 @Override