Skip to content

Commit b7c2c7d

Browse files
tmontgomerytmontgomery
committed
Merge pull request #55 from ReactiveSocket/data-metadata-buffer-capacity
move to remaining instead of capacity for data and metadata
2 parents 3205d85 + d57977d commit b7c2c7d

10 files changed

Lines changed: 53 additions & 41 deletions

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

Lines changed: 9 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -234,7 +234,7 @@ public void wrap(final int streamId, final FrameType type, final ByteBuffer data
234234
POOL.release(this.directBuffer);
235235

236236
this.directBuffer =
237-
POOL.acquireMutableDirectBuffer(FrameHeaderFlyweight.computeFrameHeaderLength(type, 0, data.capacity()));
237+
POOL.acquireMutableDirectBuffer(FrameHeaderFlyweight.computeFrameHeaderLength(type, 0, data.remaining()));
238238

239239
this.length = FrameHeaderFlyweight.encode(this.directBuffer, offset, streamId, 0, type, NULL_BYTEBUFFER, data);
240240
}
@@ -261,7 +261,7 @@ public static Frame from(
261261
final ByteBuffer data = payload.getData();
262262

263263
final Frame frame =
264-
POOL.acquireFrame(SetupFrameFlyweight.computeFrameLength(metadataMimeType, dataMimeType, metadata.capacity(), data.capacity()));
264+
POOL.acquireFrame(SetupFrameFlyweight.computeFrameLength(metadataMimeType, dataMimeType, metadata.remaining(), data.remaining()));
265265

266266
frame.length = SetupFrameFlyweight.encode(
267267
frame.directBuffer, frame.offset, flags, keepaliveInterval, maxLifetime, metadataMimeType, dataMimeType, metadata, data);
@@ -317,7 +317,7 @@ public static Frame from(
317317
) {
318318
final int code = ErrorFrameFlyweight.errorCodeFromException(throwable);
319319
final Frame frame = POOL.acquireFrame(
320-
ErrorFrameFlyweight.computeFrameLength(metadata.capacity(), data.capacity()));
320+
ErrorFrameFlyweight.computeFrameLength(metadata.remaining(), data.remaining()));
321321

322322
frame.length = ErrorFrameFlyweight.encode(
323323
frame.directBuffer, frame.offset, streamId, code, metadata, data);
@@ -354,7 +354,7 @@ public static class Lease
354354
{
355355
public static Frame from(int ttl, int numberOfRequests, ByteBuffer metadata)
356356
{
357-
final Frame frame = POOL.acquireFrame(LeaseFrameFlyweight.computeFrameLength(metadata.capacity()));
357+
final Frame frame = POOL.acquireFrame(LeaseFrameFlyweight.computeFrameLength(metadata.remaining()));
358358

359359
frame.length = LeaseFrameFlyweight.encode(frame.directBuffer, frame.offset, ttl, numberOfRequests, metadata);
360360
return frame;
@@ -397,7 +397,7 @@ public static Frame from(int streamId, FrameType type, Payload payload, int init
397397
final ByteBuffer d = payload.getData() != null ? payload.getData() : NULL_BYTEBUFFER;
398398
final ByteBuffer md = payload.getMetadata() != null ? payload.getMetadata() : NULL_BYTEBUFFER;
399399

400-
final Frame frame = POOL.acquireFrame(RequestFrameFlyweight.computeFrameLength(type, md.capacity(), d.capacity()));
400+
final Frame frame = POOL.acquireFrame(RequestFrameFlyweight.computeFrameLength(type, md.remaining(), d.remaining()));
401401

402402
if (type.hasInitialRequestN())
403403
{
@@ -421,7 +421,7 @@ public static Frame from(int streamId, FrameType type, int flags)
421421

422422
public static Frame from(int streamId, FrameType type, ByteBuffer metadata, ByteBuffer data, int initialRequestN, int flags)
423423
{
424-
final Frame frame = POOL.acquireFrame(RequestFrameFlyweight.computeFrameLength(type, metadata.capacity(), data.capacity()));
424+
final Frame frame = POOL.acquireFrame(RequestFrameFlyweight.computeFrameLength(type, metadata.remaining(), data.remaining()));
425425

426426
frame.length = RequestFrameFlyweight.encode(frame.directBuffer, frame.offset, streamId, flags, type, initialRequestN, metadata, data);
427427
return frame;
@@ -471,7 +471,7 @@ public static Frame from(int streamId, FrameType type, Payload payload)
471471
final ByteBuffer metadata = payload.getMetadata() != null ? payload.getMetadata() : NULL_BYTEBUFFER;
472472

473473
final Frame frame =
474-
POOL.acquireFrame(FrameHeaderFlyweight.computeFrameHeaderLength(type, metadata.capacity(), data.capacity()));
474+
POOL.acquireFrame(FrameHeaderFlyweight.computeFrameHeaderLength(type, metadata.remaining(), data.remaining()));
475475

476476
frame.length = FrameHeaderFlyweight.encode(frame.directBuffer, frame.offset, streamId, 0, type, metadata, data);
477477
return frame;
@@ -480,7 +480,7 @@ public static Frame from(int streamId, FrameType type, Payload payload)
480480
public static Frame from(int streamId, FrameType type, ByteBuffer metadata, ByteBuffer data, int flags)
481481
{
482482
final Frame frame =
483-
POOL.acquireFrame(FrameHeaderFlyweight.computeFrameHeaderLength(type, metadata.capacity(), data.capacity()));
483+
POOL.acquireFrame(FrameHeaderFlyweight.computeFrameHeaderLength(type, metadata.remaining(), data.remaining()));
484484

485485
frame.length = FrameHeaderFlyweight.encode(frame.directBuffer, frame.offset, streamId, flags, type, metadata, data);
486486
return frame;
@@ -515,7 +515,7 @@ public static class Keepalive
515515
public static Frame from(ByteBuffer data, boolean respond)
516516
{
517517
final Frame frame =
518-
POOL.acquireFrame(FrameHeaderFlyweight.computeFrameHeaderLength(FrameType.KEEPALIVE, 0, data.capacity()));
518+
POOL.acquireFrame(FrameHeaderFlyweight.computeFrameHeaderLength(FrameType.KEEPALIVE, 0, data.remaining()));
519519

520520
final int flags = (respond ? FrameHeaderFlyweight.FLAGS_KEEPALIVE_R : 0);
521521

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

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -56,7 +56,7 @@ public static int encode(
5656
final ByteBuffer metadata,
5757
final ByteBuffer data
5858
) {
59-
final int frameLength = computeFrameLength(metadata.capacity(), data.capacity());
59+
final int frameLength = computeFrameLength(metadata.remaining(), data.remaining());
6060

6161
int length = FrameHeaderFlyweight.encodeFrameHeader(
6262
mutableDirectBuffer, offset, frameLength, 0, FrameType.ERROR, streamId);

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

Lines changed: 8 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -110,16 +110,17 @@ public static int encodeMetadata(
110110
final ByteBuffer metadata)
111111
{
112112
int length = 0;
113+
final int metadataLength = metadata.remaining();
113114

114-
if (0 < metadata.capacity())
115+
if (0 < metadataLength)
115116
{
116117
int flags = mutableDirectBuffer.getShort(frameHeaderStartOffset + FLAGS_FIELD_OFFSET, ByteOrder.BIG_ENDIAN);
117118
flags |= FLAGS_M;
118119
mutableDirectBuffer.putShort(frameHeaderStartOffset + FLAGS_FIELD_OFFSET, (short)flags, ByteOrder.BIG_ENDIAN);
119120
mutableDirectBuffer.putInt(metadataOffset, metadata.capacity() + BitUtil.SIZE_OF_INT, ByteOrder.BIG_ENDIAN);
120121
length += BitUtil.SIZE_OF_INT;
121-
mutableDirectBuffer.putBytes(metadataOffset + length, metadata, metadata.capacity());
122-
length += metadata.capacity();
122+
mutableDirectBuffer.putBytes(metadataOffset + length, metadata, metadataLength);
123+
length += metadataLength;
123124
}
124125

125126
return length;
@@ -131,11 +132,12 @@ public static int encodeData(
131132
final ByteBuffer data)
132133
{
133134
int length = 0;
135+
final int dataLength = data.remaining();
134136

135137
if (0 < data.capacity())
136138
{
137-
mutableDirectBuffer.putBytes(dataOffset, data, data.capacity());
138-
length += data.capacity();
139+
mutableDirectBuffer.putBytes(dataOffset, data, dataLength);
140+
length += dataLength;
139141
}
140142

141143
return length;
@@ -151,7 +153,7 @@ public static int encode(
151153
final ByteBuffer metadata,
152154
final ByteBuffer data)
153155
{
154-
final int frameLength = computeFrameHeaderLength(frameType, metadata.capacity(), data.capacity());
156+
final int frameLength = computeFrameHeaderLength(frameType, metadata.remaining(), data.remaining());
155157

156158
final FrameType outFrameType;
157159

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

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -35,7 +35,7 @@ public static int encode(
3535
final int offset,
3636
final ByteBuffer data)
3737
{
38-
final int frameLength = computeFrameLength(data.capacity());
38+
final int frameLength = computeFrameLength(data.remaining());
3939

4040
int length = FrameHeaderFlyweight.encodeFrameHeader(mutableDirectBuffer, offset, frameLength, 0, FrameType.KEEPALIVE, 0);
4141

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

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -44,7 +44,7 @@ public static int encode(
4444
final int numRequests,
4545
final ByteBuffer metadata)
4646
{
47-
final int frameLength = computeFrameLength(metadata.capacity());
47+
final int frameLength = computeFrameLength(metadata.remaining());
4848

4949
int length = FrameHeaderFlyweight.encodeFrameHeader(mutableDirectBuffer, offset, frameLength, 0, FrameType.LEASE, 0);
5050

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

Lines changed: 6 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -71,14 +71,16 @@ public void append(final Payload payload)
7171
{
7272
final ByteBuffer payloadData = payload.getData();
7373
final ByteBuffer payloadMetadata = payload.getMetadata();
74+
final int dataLength = payloadData.remaining();
75+
final int metadataLength = payloadMetadata.remaining();
7476

75-
ensureDataCapacity(payloadData.capacity());
76-
ensureMetadataCapacity(payloadMetadata.capacity());
77+
ensureDataCapacity(dataLength);
78+
ensureMetadataCapacity(metadataLength);
7779

7880
dataMutableDirectBuffer.putBytes(dataLimit, payloadData, payloadData.capacity());
79-
dataLimit += payloadData.capacity();
81+
dataLimit += dataLength;
8082
metadataMutableDirectBuffer.putBytes(metadataLimit, payloadMetadata, payloadMetadata.capacity());
81-
metadataLimit += payloadMetadata.capacity();
83+
metadataLimit += metadataLength;
8284
}
8385

8486
private void ensureDataCapacity(final int additionalCapacity)

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

Lines changed: 4 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -65,13 +65,13 @@ public void resetForRequestChannel(final int streamId, final Payload payload, fi
6565

6666
public boolean hasNext()
6767
{
68-
return (dataOffset < data.capacity() || metadataOffset < metadata.capacity());
68+
return (dataOffset < data.capacity() || metadataOffset < metadata.remaining());
6969
}
7070

7171
public Frame next()
7272
{
73-
final int metadataLength = Math.min(metadataMtu, metadata.capacity() - metadataOffset);
74-
final int dataLength = Math.min(dataMtu, data.capacity() - dataOffset);
73+
final int metadataLength = Math.min(metadataMtu, metadata.remaining() - metadataOffset);
74+
final int dataLength = Math.min(dataMtu, data.remaining() - dataOffset);
7575

7676
Frame result = null;
7777

@@ -84,7 +84,7 @@ public Frame next()
8484
metadataOffset += metadataLength;
8585
dataOffset += dataLength;
8686

87-
final boolean isMoreFollowing = (metadataOffset < metadata.capacity() || dataOffset < data.capacity());
87+
final boolean isMoreFollowing = (metadataOffset < metadata.remaining() || dataOffset < data.remaining());
8888
int flags = 0;
8989

9090
if (Type.RESPONSE == type)

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

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -53,7 +53,7 @@ public static int encode(
5353
final ByteBuffer metadata,
5454
final ByteBuffer data)
5555
{
56-
final int frameLength = computeFrameLength(type, metadata.capacity(), data.capacity());
56+
final int frameLength = computeFrameLength(type, metadata.remaining(), data.remaining());
5757

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

8181
int length = FrameHeaderFlyweight.encodeFrameHeader(mutableDirectBuffer, offset, frameLength, flags, type, streamId);
8282

‎src/main/java/io/reactivesocket/internal/SetupFrameFlyweight.java‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -63,7 +63,7 @@ public static int encode(
6363
final ByteBuffer metadata,
6464
final ByteBuffer data)
6565
{
66-
final int frameLength = computeFrameLength(metadataMimeType, dataMimeType, metadata.capacity(), data.capacity());
66+
final int frameLength = computeFrameLength(metadataMimeType, dataMimeType, metadata.remaining(), data.remaining());
6767

6868
int length = FrameHeaderFlyweight.encodeFrameHeader(mutableDirectBuffer, offset, frameLength, flags, FrameType.SETUP, 0);
6969

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

Lines changed: 20 additions & 12 deletions
Original file line numberDiff line numberDiff line change
@@ -26,11 +26,19 @@ public class ThreadSafeFramePool implements FramePool
2626
{
2727
private static final int MAX_CACHED_FRAMES = 16;
2828

29-
private static final OneToOneConcurrentArrayQueue<Frame> FRAME_QUEUE =
30-
new OneToOneConcurrentArrayQueue<>(MAX_CACHED_FRAMES);
29+
private final OneToOneConcurrentArrayQueue<Frame> frameQueue;
30+
private final OneToOneConcurrentArrayQueue<MutableDirectBuffer> directBufferQueue;
3131

32-
private static final OneToOneConcurrentArrayQueue<MutableDirectBuffer> DIRECTBUFFER_QUEUE =
33-
new OneToOneConcurrentArrayQueue<>(MAX_CACHED_FRAMES);
32+
public ThreadSafeFramePool()
33+
{
34+
this(MAX_CACHED_FRAMES, MAX_CACHED_FRAMES);
35+
}
36+
37+
public ThreadSafeFramePool(final int frameQueueLength, final int directBufferQueueLength)
38+
{
39+
frameQueue = new OneToOneConcurrentArrayQueue<>(frameQueueLength);
40+
directBufferQueue = new OneToOneConcurrentArrayQueue<>(directBufferQueueLength);
41+
}
3442

3543
public Frame acquireFrame(int size)
3644
{
@@ -89,33 +97,33 @@ public MutableDirectBuffer acquireMutableDirectBuffer(int size)
8997

9098
public void release(Frame frame)
9199
{
92-
synchronized (FRAME_QUEUE)
100+
synchronized (frameQueue)
93101
{
94-
FRAME_QUEUE.offer(frame);
102+
frameQueue.offer(frame);
95103
}
96104
}
97105

98106
public void release(MutableDirectBuffer mutableDirectBuffer)
99107
{
100-
synchronized (DIRECTBUFFER_QUEUE)
108+
synchronized (directBufferQueue)
101109
{
102-
DIRECTBUFFER_QUEUE.offer(mutableDirectBuffer);
110+
directBufferQueue.offer(mutableDirectBuffer);
103111
}
104112
}
105113

106114
private Frame pollFrame()
107115
{
108-
synchronized (FRAME_QUEUE)
116+
synchronized (frameQueue)
109117
{
110-
return FRAME_QUEUE.poll();
118+
return frameQueue.poll();
111119
}
112120
}
113121

114122
private MutableDirectBuffer pollMutableDirectBuffer()
115123
{
116-
synchronized (DIRECTBUFFER_QUEUE)
124+
synchronized (directBufferQueue)
117125
{
118-
return DIRECTBUFFER_QUEUE.poll();
126+
return directBufferQueue.poll();
119127
}
120128
}
121129
}

0 commit comments

Comments
 (0)