Add CompositeByteBuf.addComponent(boolean ...) method to simplify usage

Motivation:

At the moment the user is responsible to increase the writer index of the composite buffer when a new component is added. We should add some methods that handle this for the user as this is the most popular usage of the composite buffer.

Modifications:

Add new methods that autoamtically increase the writerIndex when buffers are added.

Result:

Easier usage of CompositeByteBuf.
This commit is contained in:
Norman Maurer 2016-05-17 14:19:33 +02:00
parent 4c048d069d
commit 454cbfe651
12 changed files with 161 additions and 71 deletions

View File

@ -722,18 +722,42 @@ final class AdvancedLeakAwareCompositeByteBuf extends WrappedCompositeByteBuf {
return super.addComponents(cIndex, buffers);
}
@Override
public CompositeByteBuf removeComponent(int cIndex) {
recordLeakNonRefCountingOperation(leak);
return super.removeComponent(cIndex);
}
@Override
public CompositeByteBuf addComponents(int cIndex, Iterable<ByteBuf> buffers) {
recordLeakNonRefCountingOperation(leak);
return super.addComponents(cIndex, buffers);
}
@Override
public CompositeByteBuf addComponent(boolean increaseWriterIndex, ByteBuf buffer) {
recordLeakNonRefCountingOperation(leak);
return super.addComponent(increaseWriterIndex, buffer);
}
@Override
public CompositeByteBuf addComponents(boolean increaseWriterIndex, ByteBuf... buffers) {
recordLeakNonRefCountingOperation(leak);
return super.addComponents(increaseWriterIndex, buffers);
}
@Override
public CompositeByteBuf addComponents(boolean increaseWriterIndex, Iterable<ByteBuf> buffers) {
recordLeakNonRefCountingOperation(leak);
return super.addComponents(increaseWriterIndex, buffers);
}
@Override
public CompositeByteBuf addComponent(boolean increaseWriterIndex, int cIndex, ByteBuf buffer) {
recordLeakNonRefCountingOperation(leak);
return super.addComponent(increaseWriterIndex, cIndex, buffer);
}
@Override
public CompositeByteBuf removeComponent(int cIndex) {
recordLeakNonRefCountingOperation(leak);
return super.removeComponent(cIndex);
}
@Override
public CompositeByteBuf removeComponents(int cIndex, int numComponents) {
recordLeakNonRefCountingOperation(leak);

View File

@ -78,7 +78,7 @@ public class CompositeByteBuf extends AbstractReferenceCountedByteBuf implements
this.maxNumComponents = maxNumComponents;
components = newList(maxNumComponents);
addComponents0(0, buffers);
addComponents0(false, 0, buffers);
consolidateIfNeeded();
setIndex(0, capacity());
}
@ -99,7 +99,7 @@ public class CompositeByteBuf extends AbstractReferenceCountedByteBuf implements
this.maxNumComponents = maxNumComponents;
components = newList(maxNumComponents);
addComponents0(0, buffers);
addComponents0(false, 0, buffers);
consolidateIfNeeded();
setIndex(0, capacity());
}
@ -121,24 +121,21 @@ public class CompositeByteBuf extends AbstractReferenceCountedByteBuf implements
* Add the given {@link ByteBuf}.
* <p>
* Be aware that this method does not increase the {@code writerIndex} of the {@link CompositeByteBuf}.
* If you need to have it increased you need to handle it by your own.
* If you need to have it increased use {@link #addComponent(boolean, ByteBuf)}.
* <p>
* {@link ByteBuf#release()} ownership of {@code buffer} is transfered to this {@link CompositeByteBuf}.
* @param buffer the {@link ByteBuf} to add. {@link ByteBuf#release()} ownership is transfered to this
* {@link CompositeByteBuf}.
*/
public CompositeByteBuf addComponent(ByteBuf buffer) {
checkNotNull(buffer, "buffer");
addComponent0(components.size(), buffer);
consolidateIfNeeded();
return this;
return addComponent(false, buffer);
}
/**
* Add the given {@link ByteBuf}s.
* <p>
* Be aware that this method does not increase the {@code writerIndex} of the {@link CompositeByteBuf}.
* If you need to have it increased you need to handle it by your own.
* If you need to have it increased use {@link #addComponents(boolean, ByteBuf[])}.
* <p>
* {@link ByteBuf#release()} ownership of all {@link ByteBuf} objects in {@code buffers} is transfered to this
* {@link CompositeByteBuf}.
@ -146,16 +143,14 @@ public class CompositeByteBuf extends AbstractReferenceCountedByteBuf implements
* ownership of all {@link ByteBuf} objects is transfered to this {@link CompositeByteBuf}.
*/
public CompositeByteBuf addComponents(ByteBuf... buffers) {
addComponents0(components.size(), buffers);
consolidateIfNeeded();
return this;
return addComponents(false, buffers);
}
/**
* Add the given {@link ByteBuf}s.
* <p>
* Be aware that this method does not increase the {@code writerIndex} of the {@link CompositeByteBuf}.
* If you need to have it increased you need to handle it by your own.
* If you need to have it increased use {@link #addComponents(boolean, Iterable)}.
* <p>
* {@link ByteBuf#release()} ownership of all {@link ByteBuf} objects in {@code buffers} is transfered to this
* {@link CompositeByteBuf}.
@ -163,16 +158,14 @@ public class CompositeByteBuf extends AbstractReferenceCountedByteBuf implements
* ownership of all {@link ByteBuf} objects is transfered to this {@link CompositeByteBuf}.
*/
public CompositeByteBuf addComponents(Iterable<ByteBuf> buffers) {
addComponents0(components.size(), buffers);
consolidateIfNeeded();
return this;
return addComponents(false, buffers);
}
/**
* Add the given {@link ByteBuf} on the specific index.
* <p>
* Be aware that this method does not increase the {@code writerIndex} of the {@link CompositeByteBuf}.
* If you need to have it increased you need to handle it by your own.
* If you need to have it increased use {@link #addComponent(boolean, int, ByteBuf)}.
* <p>
* {@link ByteBuf#release()} ownership of {@code buffer} is transfered to this {@link CompositeByteBuf}.
* @param cIndex the index on which the {@link ByteBuf} will be added.
@ -180,8 +173,66 @@ public class CompositeByteBuf extends AbstractReferenceCountedByteBuf implements
* {@link CompositeByteBuf}.
*/
public CompositeByteBuf addComponent(int cIndex, ByteBuf buffer) {
return addComponent(false, cIndex, buffer);
}
/**
* Add the given {@link ByteBuf} and increase the {@code writerIndex} if {@code increaseWriterIndex} is
* {@code true}.
*
* {@link ByteBuf#release()} ownership of {@code buffer} is transfered to this {@link CompositeByteBuf}.
* @param buffer the {@link ByteBuf} to add. {@link ByteBuf#release()} ownership is transfered to this
* {@link CompositeByteBuf}.
*/
public CompositeByteBuf addComponent(boolean increaseWriterIndex, ByteBuf buffer) {
checkNotNull(buffer, "buffer");
addComponent0(cIndex, buffer);
addComponent0(increaseWriterIndex, components.size(), buffer);
consolidateIfNeeded();
return this;
}
/**
* Add the given {@link ByteBuf}s and increase the {@code writerIndex} if {@code increaseWriterIndex} is
* {@code true}.
*
* {@link ByteBuf#release()} ownership of all {@link ByteBuf} objects in {@code buffers} is transfered to this
* {@link CompositeByteBuf}.
* @param buffers the {@link ByteBuf}s to add. {@link ByteBuf#release()} ownership of all {@link ByteBuf#release()}
* ownership of all {@link ByteBuf} objects is transfered to this {@link CompositeByteBuf}.
*/
public CompositeByteBuf addComponents(boolean increaseWriterIndex, ByteBuf... buffers) {
addComponents0(increaseWriterIndex, components.size(), buffers);
consolidateIfNeeded();
return this;
}
/**
* Add the given {@link ByteBuf}s and increase the {@code writerIndex} if {@code increaseWriterIndex} is
* {@code true}.
*
* {@link ByteBuf#release()} ownership of all {@link ByteBuf} objects in {@code buffers} is transfered to this
* {@link CompositeByteBuf}.
* @param buffers the {@link ByteBuf}s to add. {@link ByteBuf#release()} ownership of all {@link ByteBuf#release()}
* ownership of all {@link ByteBuf} objects is transfered to this {@link CompositeByteBuf}.
*/
public CompositeByteBuf addComponents(boolean increaseWriterIndex, Iterable<ByteBuf> buffers) {
addComponents0(increaseWriterIndex, components.size(), buffers);
consolidateIfNeeded();
return this;
}
/**
* Add the given {@link ByteBuf} on the specific index and increase the {@code writerIndex}
* if {@code increaseWriterIndex} is {@code true}.
*
* {@link ByteBuf#release()} ownership of {@code buffer} is transfered to this {@link CompositeByteBuf}.
* @param cIndex the index on which the {@link ByteBuf} will be added.
* @param buffer the {@link ByteBuf} to add. {@link ByteBuf#release()} ownership is transfered to this
* {@link CompositeByteBuf}.
*/
public CompositeByteBuf addComponent(boolean increaseWriterIndex, int cIndex, ByteBuf buffer) {
checkNotNull(buffer, "buffer");
addComponent0(increaseWriterIndex, cIndex, buffer);
consolidateIfNeeded();
return this;
}
@ -189,7 +240,7 @@ public class CompositeByteBuf extends AbstractReferenceCountedByteBuf implements
/**
* Precondition is that {@code buffer != null}.
*/
private int addComponent0(int cIndex, ByteBuf buffer) {
private int addComponent0(boolean increaseWriterIndex, int cIndex, ByteBuf buffer) {
assert buffer != null;
boolean wasAdded = false;
try {
@ -216,6 +267,9 @@ public class CompositeByteBuf extends AbstractReferenceCountedByteBuf implements
updateComponentOffsets(cIndex);
}
}
if (increaseWriterIndex) {
writerIndex(writerIndex() + buffer.readableBytes());
}
return cIndex;
} finally {
if (!wasAdded) {
@ -239,12 +293,12 @@ public class CompositeByteBuf extends AbstractReferenceCountedByteBuf implements
* ownership of all {@link ByteBuf} objects is transfered to this {@link CompositeByteBuf}.
*/
public CompositeByteBuf addComponents(int cIndex, ByteBuf... buffers) {
addComponents0(cIndex, buffers);
addComponents0(false, cIndex, buffers);
consolidateIfNeeded();
return this;
}
private int addComponents0(int cIndex, ByteBuf... buffers) {
private int addComponents0(boolean increaseWriterIndex, int cIndex, ByteBuf... buffers) {
checkNotNull(buffers, "buffers");
int i = 0;
try {
@ -258,7 +312,7 @@ public class CompositeByteBuf extends AbstractReferenceCountedByteBuf implements
if (b == null) {
break;
}
cIndex = addComponent0(cIndex, b) + 1;
cIndex = addComponent0(increaseWriterIndex, cIndex, b) + 1;
int size = components.size();
if (cIndex > size) {
cIndex = size;
@ -293,15 +347,15 @@ public class CompositeByteBuf extends AbstractReferenceCountedByteBuf implements
* {@link CompositeByteBuf}.
*/
public CompositeByteBuf addComponents(int cIndex, Iterable<ByteBuf> buffers) {
addComponents0(cIndex, buffers);
addComponents0(false, cIndex, buffers);
consolidateIfNeeded();
return this;
}
private int addComponents0(int cIndex, Iterable<ByteBuf> buffers) {
private int addComponents0(boolean increaseIndex, int cIndex, Iterable<ByteBuf> buffers) {
if (buffers instanceof ByteBuf) {
// If buffers also implements ByteBuf (e.g. CompositeByteBuf), it has to go to addComponent(ByteBuf).
return addComponent0(cIndex, (ByteBuf) buffers);
return addComponent0(increaseIndex, cIndex, (ByteBuf) buffers);
}
checkNotNull(buffers, "buffers");
@ -328,7 +382,7 @@ public class CompositeByteBuf extends AbstractReferenceCountedByteBuf implements
}
Collection<ByteBuf> col = (Collection<ByteBuf>) buffers;
return addComponents0(cIndex, col.toArray(new ByteBuf[col.size()]));
return addComponents0(increaseIndex, cIndex, col.toArray(new ByteBuf[col.size()]));
}
/**
@ -595,13 +649,13 @@ public class CompositeByteBuf extends AbstractReferenceCountedByteBuf implements
if (nComponents < maxNumComponents) {
padding = allocBuffer(paddingLength);
padding.setIndex(0, paddingLength);
addComponent0(components.size(), padding);
addComponent0(false, components.size(), padding);
} else {
padding = allocBuffer(paddingLength);
padding.setIndex(0, paddingLength);
// FIXME: No need to create a padding buffer and consolidate.
// Just create a big single buffer and put the current content there.
addComponent0(components.size(), padding);
addComponent0(false, components.size(), padding);
consolidateIfNeeded();
}
} else if (newCapacity < oldCapacity) {

View File

@ -386,6 +386,30 @@ class WrappedCompositeByteBuf extends CompositeByteBuf {
return this;
}
@Override
public CompositeByteBuf addComponent(boolean increaseWriterIndex, ByteBuf buffer) {
wrapped.addComponent(increaseWriterIndex, buffer);
return this;
}
@Override
public CompositeByteBuf addComponents(boolean increaseWriterIndex, ByteBuf... buffers) {
wrapped.addComponents(increaseWriterIndex, buffers);
return this;
}
@Override
public CompositeByteBuf addComponents(boolean increaseWriterIndex, Iterable<ByteBuf> buffers) {
wrapped.addComponents(increaseWriterIndex, buffers);
return this;
}
@Override
public CompositeByteBuf addComponent(boolean increaseWriterIndex, int cIndex, ByteBuf buffer) {
wrapped.addComponent(increaseWriterIndex, cIndex, buffer);
return this;
}
@Override
public CompositeByteBuf removeComponent(int cIndex) {
wrapped.removeComponent(cIndex);

View File

@ -846,11 +846,11 @@ public abstract class AbstractCompositeByteBufTest extends AbstractByteBufTest {
public void testAddEmptyBufferInMiddle() {
CompositeByteBuf cbuf = compositeBuffer();
ByteBuf buf1 = buffer().writeByte((byte) 1);
cbuf.addComponent(buf1).writerIndex(cbuf.writerIndex() + buf1.readableBytes());
cbuf.addComponent(true, buf1);
ByteBuf buf2 = EMPTY_BUFFER;
cbuf.addComponent(buf2).writerIndex(cbuf.writerIndex() + buf2.readableBytes());
cbuf.addComponent(true, buf2);
ByteBuf buf3 = buffer().writeByte((byte) 2);
cbuf.addComponent(buf3).writerIndex(cbuf.writerIndex() + buf3.readableBytes());
cbuf.addComponent(true, buf3);
assertEquals(2, cbuf.readableBytes());
assertEquals((byte) 1, cbuf.readByte());

View File

@ -100,12 +100,10 @@ public abstract class AbstractMemoryHttpData extends AbstractHttpData {
byteBuf = buffer;
} else if (byteBuf instanceof CompositeByteBuf) {
CompositeByteBuf cbb = (CompositeByteBuf) byteBuf;
cbb.addComponent(buffer);
cbb.writerIndex(cbb.writerIndex() + buffer.readableBytes());
cbb.addComponent(true, buffer);
} else {
CompositeByteBuf cbb = compositeBuffer(Integer.MAX_VALUE);
cbb.addComponents(byteBuf, buffer);
cbb.writerIndex(byteBuf.readableBytes() + buffer.readableBytes());
cbb.addComponents(true, byteBuf, buffer);
byteBuf = cbb;
}
}

View File

@ -103,15 +103,13 @@ public abstract class WebSocketClientHandshakerTest {
byte[] bytes = "HTTP/1.1 101 Switching Protocols\r\nContent-Length: 0\r\n\r\n".getBytes(CharsetUtil.US_ASCII);
CompositeByteBuf compositeByteBuf = Unpooled.compositeBuffer();
compositeByteBuf.addComponent(Unpooled.wrappedBuffer(bytes));
compositeByteBuf.writerIndex(compositeByteBuf.writerIndex() + bytes.length);
compositeByteBuf.addComponent(true, Unpooled.wrappedBuffer(bytes));
for (;;) {
ByteBuf frameBytes = (ByteBuf) websocketChannel.readOutbound();
if (frameBytes == null) {
break;
}
compositeByteBuf.addComponent(frameBytes);
compositeByteBuf.writerIndex(compositeByteBuf.writerIndex() + frameBytes.readableBytes());
compositeByteBuf.addComponent(true, frameBytes);
}
EmbeddedChannel ch = new EmbeddedChannel(new HttpObjectAggregator(Integer.MAX_VALUE),

View File

@ -119,11 +119,10 @@ public abstract class ByteToMessageDecoder extends ChannelInboundHandlerAdapter
if (cumulation instanceof CompositeByteBuf) {
composite = (CompositeByteBuf) cumulation;
} else {
int readable = cumulation.readableBytes();
composite = alloc.compositeBuffer(Integer.MAX_VALUE);
composite.addComponent(cumulation).writerIndex(readable);
composite.addComponent(true, cumulation);
}
composite.addComponent(in).writerIndex(composite.writerIndex() + in.readableBytes());
composite.addComponent(true, in);
buffer = composite;
}
return buffer;

View File

@ -94,8 +94,7 @@ public class SnappyFramedEncoderTest {
if (m == null) {
break;
}
actual.addComponent(m);
actual.writerIndex(actual.writerIndex() + m.readableBytes());
actual.addComponent(true, m);
}
assertEquals(releaseLater(expected), releaseLater(actual));
in.release();

View File

@ -63,15 +63,13 @@ public class DatagramUnicastTest extends AbstractDatagramTest {
public void testSimpleSendCompositeDirectByteBuf(Bootstrap sb, Bootstrap cb) throws Throwable {
CompositeByteBuf buf = Unpooled.compositeBuffer();
buf.addComponent(Unpooled.directBuffer().writeBytes(BYTES, 0, 2));
buf.addComponent(Unpooled.directBuffer().writeBytes(BYTES, 2, 2));
buf.writerIndex(4);
buf.addComponent(true, Unpooled.directBuffer().writeBytes(BYTES, 0, 2));
buf.addComponent(true, Unpooled.directBuffer().writeBytes(BYTES, 2, 2));
testSimpleSend0(sb, cb, buf, true, BYTES, 1);
CompositeByteBuf buf2 = Unpooled.compositeBuffer();
buf2.addComponent(Unpooled.directBuffer().writeBytes(BYTES, 0, 2));
buf2.addComponent(Unpooled.directBuffer().writeBytes(BYTES, 2, 2));
buf2.writerIndex(4);
buf2.addComponent(true, Unpooled.directBuffer().writeBytes(BYTES, 0, 2));
buf2.addComponent(true, Unpooled.directBuffer().writeBytes(BYTES, 2, 2));
testSimpleSend0(sb, cb, buf2, true, BYTES, 4);
}
@ -82,15 +80,13 @@ public class DatagramUnicastTest extends AbstractDatagramTest {
public void testSimpleSendCompositeHeapByteBuf(Bootstrap sb, Bootstrap cb) throws Throwable {
CompositeByteBuf buf = Unpooled.compositeBuffer();
buf.addComponent(Unpooled.buffer().writeBytes(BYTES, 0, 2));
buf.addComponent(Unpooled.buffer().writeBytes(BYTES, 2, 2));
buf.writerIndex(4);
buf.addComponent(true, Unpooled.buffer().writeBytes(BYTES, 0, 2));
buf.addComponent(true, Unpooled.buffer().writeBytes(BYTES, 2, 2));
testSimpleSend0(sb, cb, buf, true, BYTES, 1);
CompositeByteBuf buf2 = Unpooled.compositeBuffer();
buf2.addComponent(Unpooled.buffer().writeBytes(BYTES, 0, 2));
buf2.addComponent(Unpooled.buffer().writeBytes(BYTES, 2, 2));
buf2.writerIndex(4);
buf2.addComponent(true, Unpooled.buffer().writeBytes(BYTES, 0, 2));
buf2.addComponent(true, Unpooled.buffer().writeBytes(BYTES, 2, 2));
testSimpleSend0(sb, cb, buf2, true, BYTES, 4);
}
@ -101,15 +97,13 @@ public class DatagramUnicastTest extends AbstractDatagramTest {
public void testSimpleSendCompositeMixedByteBuf(Bootstrap sb, Bootstrap cb) throws Throwable {
CompositeByteBuf buf = Unpooled.compositeBuffer();
buf.addComponent(Unpooled.directBuffer().writeBytes(BYTES, 0, 2));
buf.addComponent(Unpooled.buffer().writeBytes(BYTES, 2, 2));
buf.writerIndex(4);
buf.addComponent(true, Unpooled.directBuffer().writeBytes(BYTES, 0, 2));
buf.addComponent(true, Unpooled.buffer().writeBytes(BYTES, 2, 2));
testSimpleSend0(sb, cb, buf, true, BYTES, 1);
CompositeByteBuf buf2 = Unpooled.compositeBuffer();
buf2.addComponent(Unpooled.directBuffer().writeBytes(BYTES, 0, 2));
buf2.addComponent(Unpooled.buffer().writeBytes(BYTES, 2, 2));
buf2.writerIndex(4);
buf2.addComponent(true, Unpooled.directBuffer().writeBytes(BYTES, 0, 2));
buf2.addComponent(true, Unpooled.buffer().writeBytes(BYTES, 2, 2));
testSimpleSend0(sb, cb, buf2, true, BYTES, 4);
}

View File

@ -127,7 +127,7 @@ public class SocketGatheringWriteTest extends AbstractSocketTest {
buf.writerIndex(split);
ByteBuf buf2 = Unpooled.buffer(size).writeBytes(buf, split, oldIndex - split);
CompositeByteBuf comp = Unpooled.compositeBuffer();
comp.addComponent(buf).addComponent(buf2).writerIndex(length);
comp.addComponent(true, buf).addComponent(true, buf2);
cc.write(comp);
} else {
cc.write(buf);

View File

@ -287,7 +287,7 @@ public class SocketSslEchoTest extends AbstractSocketTest {
int length = Math.min(random.nextInt(1024 * 64), data.length - clientSendCounterVal);
ByteBuf buf = Unpooled.wrappedBuffer(data, clientSendCounterVal, length);
if (useCompositeByteBuf) {
buf = Unpooled.compositeBuffer().addComponent(buf).writerIndex(buf.writerIndex());
buf = Unpooled.compositeBuffer().addComponent(true, buf);
}
ChannelFuture future = clientChannel.writeAndFlush(buf);
@ -520,7 +520,7 @@ public class SocketSslEchoTest extends AbstractSocketTest {
ByteBuf buf = Unpooled.wrappedBuffer(actual);
if (useCompositeByteBuf) {
buf = Unpooled.compositeBuffer().addComponent(buf).writerIndex(buf.writerIndex());
buf = Unpooled.compositeBuffer().addComponent(true, buf);
}
ctx.write(buf);

View File

@ -99,7 +99,7 @@ public class ChannelOutboundBufferTest {
CompositeByteBuf comp = compositeBuffer(256);
ByteBuf buf = directBuffer().writeBytes("buf1".getBytes(CharsetUtil.US_ASCII));
for (int i = 0; i < 65; i++) {
comp.addComponent(buf.copy()).writerIndex(comp.writerIndex() + buf.readableBytes());
comp.addComponent(true, buf.copy());
}
buffer.addMessage(comp, comp.readableBytes(), channel.voidPromise());