Merge pull request #54 from ReactiveSocket/frame-pool-cleanup

cleanup of some frame pooling

Author: tmontgomery

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

@@ -317,7 +317,7 @@ public static Frame from(
         ) {
             final int code = ErrorFrameFlyweight.errorCodeFromException(throwable);
             final Frame frame = POOL.acquireFrame(
-                ErrorFrameFlyweight.computeFrameLength(data.capacity(), metadata.capacity()));
+                ErrorFrameFlyweight.computeFrameLength(metadata.capacity(), data.capacity()));
 
             frame.length = ErrorFrameFlyweight.encode(
                 frame.directBuffer, frame.offset, streamId, code, metadata, data);
@@ -576,6 +576,6 @@ public String toString() {
         } catch (Exception e) {
             e.printStackTrace();
         }
-        return "Frame[" + offset + "] => Stream ID: " + streamId + " Type: " + type + " Payload Data: " + payload;
+        return "Frame[" + offset + "] => Stream ID: " + streamId + " Type: " + type + " Payload: " + payload;
     }
 }

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

@@ -40,8 +40,8 @@ public class ErrorFrameFlyweight {
     private static final int PAYLOAD_OFFSET = ERROR_CODE_FIELD_OFFSET + BitUtil.SIZE_OF_INT;
 
     public static int computeFrameLength(
-        final int dataLength,
-        final int metadataLength
+        final int metadataLength,
+        final int dataLength
     ) {
         int length = FrameHeaderFlyweight.computeFrameHeaderLength(
             FrameType.ERROR, metadataLength, dataLength);
@@ -56,7 +56,7 @@ public static int encode(
         final ByteBuffer metadata,
         final ByteBuffer data
     ) {
-        final int frameLength = computeFrameLength(data.capacity(), metadata.capacity());
+        final int frameLength = computeFrameLength(metadata.capacity(), data.capacity());
 
         int length = FrameHeaderFlyweight.encodeFrameHeader(
             mutableDirectBuffer, offset, frameLength, 0, FrameType.ERROR, streamId);
@@ -98,6 +98,6 @@ public static int errorCode(final DirectBuffer directBuffer, final int offset) {
     }
 
     public static int payloadOffset(final DirectBuffer directBuffer, final int offset) {
-        return offset + PAYLOAD_OFFSET;
+        return offset + FrameHeaderFlyweight.FRAME_HEADER_LENGTH + BitUtil.SIZE_OF_INT;
     }
 }

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

@@ -75,7 +75,7 @@ public MutableDirectBuffer acquireMutableDirectBuffer(ByteBuffer byteBuffer)
     public MutableDirectBuffer acquireMutableDirectBuffer(int size)
     {
         UnsafeBuffer directBuffer = (UnsafeBuffer)pollMutableDirectBuffer();
-        if (null == directBuffer || directBuffer.byteBuffer().capacity() < size)
+        if (null == directBuffer || directBuffer.capacity() < size)
         {
             directBuffer = new UnsafeBuffer(ByteBuffer.allocate(size));
         }
@@ -87,23 +87,35 @@ public MutableDirectBuffer acquireMutableDirectBuffer(int size)
         return directBuffer;
     }
 
-    public synchronized void release(Frame frame)
+    public void release(Frame frame)
     {
-        FRAME_QUEUE.offer(frame);
+        synchronized (FRAME_QUEUE)
+        {
+            FRAME_QUEUE.offer(frame);
+        }
     }
 
-    public synchronized void release(MutableDirectBuffer mutableDirectBuffer)
+    public void release(MutableDirectBuffer mutableDirectBuffer)
     {
-        DIRECTBUFFER_QUEUE.offer(mutableDirectBuffer);
+        synchronized (DIRECTBUFFER_QUEUE)
+        {
+            DIRECTBUFFER_QUEUE.offer(mutableDirectBuffer);
+        }
     }
 
-    private synchronized Frame pollFrame()
+    private Frame pollFrame()
     {
-        return FRAME_QUEUE.poll();
+        synchronized (FRAME_QUEUE)
+        {
+            return FRAME_QUEUE.poll();
+        }
     }
 
-    private synchronized MutableDirectBuffer pollMutableDirectBuffer()
+    private MutableDirectBuffer pollMutableDirectBuffer()
     {
-        return DIRECTBUFFER_QUEUE.poll();
+        synchronized (DIRECTBUFFER_QUEUE)
+        {
+            return DIRECTBUFFER_QUEUE.poll();
+        }
     }
 }

src/test/java/io/reactivesocket/TestConnection.java

@@ -84,13 +84,13 @@ public void connectToServerConnection(TestConnection serverConnection, boolean l
 		// client to server
 		write.add(f -> {
 			serverThread.schedule(() -> {
-				serverConnection.toInput.send(f);	
+				serverConnection.toInput.send(f);
 			});
 		});
 		// server to client
 		serverConnection.write.add(f -> {
 			clientThread.schedule(() -> {
-				toInput.send(f);	
+				toInput.send(f);
 			});
 		});
 	}