Merge branch 'master' of https://github.com/ReactiveSocket/reactivesocket-java

Author: tmontgomery

src/main/java/io/reactivesocket/Frame.java

@@ -234,7 +234,7 @@ public void wrap(final int streamId, final FrameType type, final ByteBuffer data
         POOL.release(this.directBuffer);
 
         this.directBuffer =
-            POOL.acquireMutableDirectBuffer(FrameHeaderFlyweight.computeFrameHeaderLength(type, 0, data.capacity()));
+            POOL.acquireMutableDirectBuffer(FrameHeaderFlyweight.computeFrameHeaderLength(type, 0, data.remaining()));
 
         this.length = FrameHeaderFlyweight.encode(this.directBuffer, offset, streamId, 0, type, NULL_BYTEBUFFER, data);
     }
@@ -261,7 +261,7 @@ public static Frame from(
             final ByteBuffer data = payload.getData();
 
             final Frame frame =
-                POOL.acquireFrame(SetupFrameFlyweight.computeFrameLength(metadataMimeType, dataMimeType, metadata.capacity(), data.capacity()));
+                POOL.acquireFrame(SetupFrameFlyweight.computeFrameLength(metadataMimeType, dataMimeType, metadata.remaining(), data.remaining()));
 
             frame.length = SetupFrameFlyweight.encode(
                 frame.directBuffer, frame.offset, flags, keepaliveInterval, maxLifetime, metadataMimeType, dataMimeType, metadata, data);
@@ -317,7 +317,7 @@ public static Frame from(
         ) {
             final int code = ErrorFrameFlyweight.errorCodeFromException(throwable);
             final Frame frame = POOL.acquireFrame(
-                ErrorFrameFlyweight.computeFrameLength(metadata.capacity(), data.capacity()));
+                ErrorFrameFlyweight.computeFrameLength(metadata.remaining(), data.remaining()));
 
             frame.length = ErrorFrameFlyweight.encode(
                 frame.directBuffer, frame.offset, streamId, code, metadata, data);
@@ -354,7 +354,7 @@ public static class Lease
     {
         public static Frame from(int ttl, int numberOfRequests, ByteBuffer metadata)
         {
-            final Frame frame = POOL.acquireFrame(LeaseFrameFlyweight.computeFrameLength(metadata.capacity()));
+            final Frame frame = POOL.acquireFrame(LeaseFrameFlyweight.computeFrameLength(metadata.remaining()));
 
             frame.length = LeaseFrameFlyweight.encode(frame.directBuffer, frame.offset, ttl, numberOfRequests, metadata);
             return frame;
@@ -397,7 +397,7 @@ public static Frame from(int streamId, FrameType type, Payload payload, int init
             final ByteBuffer d = payload.getData() != null ? payload.getData() : NULL_BYTEBUFFER;
             final ByteBuffer md = payload.getMetadata() != null ? payload.getMetadata() : NULL_BYTEBUFFER;
 
-            final Frame frame = POOL.acquireFrame(RequestFrameFlyweight.computeFrameLength(type, md.capacity(), d.capacity()));
+            final Frame frame = POOL.acquireFrame(RequestFrameFlyweight.computeFrameLength(type, md.remaining(), d.remaining()));
 
             if (type.hasInitialRequestN())
             {
@@ -421,7 +421,7 @@ public static Frame from(int streamId, FrameType type, int flags)
 
         public static Frame from(int streamId, FrameType type, ByteBuffer metadata, ByteBuffer data, int initialRequestN, int flags)
         {
-            final Frame frame = POOL.acquireFrame(RequestFrameFlyweight.computeFrameLength(type, metadata.capacity(), data.capacity()));
+            final Frame frame = POOL.acquireFrame(RequestFrameFlyweight.computeFrameLength(type, metadata.remaining(), data.remaining()));
 
             frame.length = RequestFrameFlyweight.encode(frame.directBuffer, frame.offset, streamId, flags, type, initialRequestN, metadata, data);
             return frame;
@@ -471,7 +471,7 @@ public static Frame from(int streamId, FrameType type, Payload payload)
             final ByteBuffer metadata = payload.getMetadata() != null ? payload.getMetadata() : NULL_BYTEBUFFER;
 
             final Frame frame =
-                POOL.acquireFrame(FrameHeaderFlyweight.computeFrameHeaderLength(type, metadata.capacity(), data.capacity()));
+                POOL.acquireFrame(FrameHeaderFlyweight.computeFrameHeaderLength(type, metadata.remaining(), data.remaining()));
 
             frame.length = FrameHeaderFlyweight.encode(frame.directBuffer, frame.offset, streamId, 0, type, metadata, data);
             return frame;
@@ -480,7 +480,7 @@ public static Frame from(int streamId, FrameType type, Payload payload)
         public static Frame from(int streamId, FrameType type, ByteBuffer metadata, ByteBuffer data, int flags)
         {
             final Frame frame =
-                POOL.acquireFrame(FrameHeaderFlyweight.computeFrameHeaderLength(type, metadata.capacity(), data.capacity()));
+                POOL.acquireFrame(FrameHeaderFlyweight.computeFrameHeaderLength(type, metadata.remaining(), data.remaining()));
 
             frame.length = FrameHeaderFlyweight.encode(frame.directBuffer, frame.offset, streamId, flags, type, metadata, data);
             return frame;
@@ -515,7 +515,7 @@ public static class Keepalive
         public static Frame from(ByteBuffer data, boolean respond)
         {
             final Frame frame =
-                POOL.acquireFrame(FrameHeaderFlyweight.computeFrameHeaderLength(FrameType.KEEPALIVE, 0, data.capacity()));
+                POOL.acquireFrame(FrameHeaderFlyweight.computeFrameHeaderLength(FrameType.KEEPALIVE, 0, data.remaining()));
 
             final int flags = (respond ? FrameHeaderFlyweight.FLAGS_KEEPALIVE_R : 0);
 

src/main/java/io/reactivesocket/internal/ErrorFrameFlyweight.java

@@ -56,7 +56,7 @@ public static int encode(
         final ByteBuffer metadata,
         final ByteBuffer data
     ) {
-        final int frameLength = computeFrameLength(metadata.capacity(), data.capacity());
+        final int frameLength = computeFrameLength(metadata.remaining(), data.remaining());
 
         int length = FrameHeaderFlyweight.encodeFrameHeader(
             mutableDirectBuffer, offset, frameLength, 0, FrameType.ERROR, streamId);

src/main/java/io/reactivesocket/internal/FrameHeaderFlyweight.java

@@ -110,16 +110,17 @@ public static int encodeMetadata(
         final ByteBuffer metadata)
     {
         int length = 0;
+        final int metadataLength = metadata.remaining();
 
-        if (0 < metadata.capacity())
+        if (0 < metadataLength)
         {
             int flags = mutableDirectBuffer.getShort(frameHeaderStartOffset + FLAGS_FIELD_OFFSET, ByteOrder.BIG_ENDIAN);
             flags |= FLAGS_M;
             mutableDirectBuffer.putShort(frameHeaderStartOffset + FLAGS_FIELD_OFFSET, (short)flags, ByteOrder.BIG_ENDIAN);
             mutableDirectBuffer.putInt(metadataOffset, metadata.capacity() + BitUtil.SIZE_OF_INT, ByteOrder.BIG_ENDIAN);
             length += BitUtil.SIZE_OF_INT;
-            mutableDirectBuffer.putBytes(metadataOffset + length, metadata, metadata.capacity());
-            length += metadata.capacity();
+            mutableDirectBuffer.putBytes(metadataOffset + length, metadata, metadataLength);
+            length += metadataLength;
         }
 
         return length;
@@ -131,11 +132,12 @@ public static int encodeData(
         final ByteBuffer data)
     {
         int length = 0;
+        final int dataLength = data.remaining();
 
         if (0 < data.capacity())
         {
-            mutableDirectBuffer.putBytes(dataOffset, data, data.capacity());
-            length += data.capacity();
+            mutableDirectBuffer.putBytes(dataOffset, data, dataLength);
+            length += dataLength;
         }
 
         return length;
@@ -151,7 +153,7 @@ public static int encode(
         final ByteBuffer metadata,
         final ByteBuffer data)
     {
-        final int frameLength = computeFrameHeaderLength(frameType, metadata.capacity(), data.capacity());
+        final int frameLength = computeFrameHeaderLength(frameType, metadata.remaining(), data.remaining());
 
         final FrameType outFrameType;
 

src/main/java/io/reactivesocket/internal/KeepaliveFrameFlyweight.java

@@ -35,7 +35,7 @@ public static int encode(
         final int offset,
         final ByteBuffer data)
     {
-        final int frameLength = computeFrameLength(data.capacity());
+        final int frameLength = computeFrameLength(data.remaining());
 
         int length = FrameHeaderFlyweight.encodeFrameHeader(mutableDirectBuffer, offset, frameLength, 0, FrameType.KEEPALIVE, 0);
 

src/main/java/io/reactivesocket/internal/LeaseFrameFlyweight.java

@@ -44,7 +44,7 @@ public static int encode(
         final int numRequests,
         final ByteBuffer metadata)
     {
-        final int frameLength = computeFrameLength(metadata.capacity());
+        final int frameLength = computeFrameLength(metadata.remaining());
 
         int length = FrameHeaderFlyweight.encodeFrameHeader(mutableDirectBuffer, offset, frameLength, 0, FrameType.LEASE, 0);
 

src/main/java/io/reactivesocket/internal/PayloadBuilder.java

@@ -71,14 +71,16 @@ public void append(final Payload payload)
     {
         final ByteBuffer payloadData = payload.getData();
         final ByteBuffer payloadMetadata = payload.getMetadata();
+        final int dataLength = payloadData.remaining();
+        final int metadataLength = payloadMetadata.remaining();
 
-        ensureDataCapacity(payloadData.capacity());
-        ensureMetadataCapacity(payloadMetadata.capacity());
+        ensureDataCapacity(dataLength);
+        ensureMetadataCapacity(metadataLength);
 
         dataMutableDirectBuffer.putBytes(dataLimit, payloadData, payloadData.capacity());
-        dataLimit += payloadData.capacity();
+        dataLimit += dataLength;
         metadataMutableDirectBuffer.putBytes(metadataLimit, payloadMetadata, payloadMetadata.capacity());
-        metadataLimit += payloadMetadata.capacity();
+        metadataLimit += metadataLength;
     }
 
     private void ensureDataCapacity(final int additionalCapacity)

src/main/java/io/reactivesocket/internal/PayloadFragmenter.java

@@ -84,13 +84,13 @@ public Iterator<Frame> iterator()
 
     public boolean hasNext()
     {
-        return (dataOffset < data.capacity() || metadataOffset < metadata.capacity());
+        return (dataOffset < data.capacity() || metadataOffset < metadata.remaining());
     }
 
     public Frame next()
     {
-        final int metadataLength = Math.min(metadataMtu, metadata.capacity() - metadataOffset);
-        final int dataLength = Math.min(dataMtu, data.capacity() - dataOffset);
+        final int metadataLength = Math.min(metadataMtu, metadata.remaining() - metadataOffset);
+        final int dataLength = Math.min(dataMtu, data.remaining() - dataOffset);
 
         Frame result = null;
 
@@ -103,7 +103,7 @@ public Frame next()
         metadataOffset += metadataLength;
         dataOffset += dataLength;
 
-        final boolean isMoreFollowing = (metadataOffset < metadata.capacity() || dataOffset < data.capacity());
+        final boolean isMoreFollowing = (metadataOffset < metadata.remaining() || dataOffset < data.remaining());
         int flags = 0;
 
         if (Type.RESPONSE == type)

src/main/java/io/reactivesocket/internal/RequestFrameFlyweight.java

@@ -53,7 +53,7 @@ public static int encode(
         final ByteBuffer metadata,
         final ByteBuffer data)
     {
-        final int frameLength = computeFrameLength(type, metadata.capacity(), data.capacity());
+        final int frameLength = computeFrameLength(type, metadata.remaining(), data.remaining());
 
         flags |= FLAGS_REQUEST_CHANNEL_N;
         int length = FrameHeaderFlyweight.encodeFrameHeader(mutableDirectBuffer, offset, frameLength, flags, type, streamId);
@@ -76,7 +76,7 @@ public static int encode(
         final ByteBuffer metadata,
         final ByteBuffer data)
     {
-        final int frameLength = computeFrameLength(type, metadata.capacity(), data.capacity());
+        final int frameLength = computeFrameLength(type, metadata.remaining(), data.remaining());
 
         int length = FrameHeaderFlyweight.encodeFrameHeader(mutableDirectBuffer, offset, frameLength, flags, type, streamId);