Add an alternative message decoder implementation
This one is a rewrite, making use of the new APIs where possible. The test uses bifurcate to cut buffers into segments.
This commit is contained in:
Chris Vest
parent
c081c73885
commit
0748d206d2
@ -0,0 +1,115 @@
|
||||
/*
|
||||
* Copyright 2021 The Netty Project
|
||||
*
|
||||
* The Netty Project licenses this file to you under the Apache License,
|
||||
* version 2.0 (the "License"); you may not use this file except in compliance
|
||||
* with the License. You may obtain a copy of the License at:
|
||||
*
|
||||
* https://www.apache.org/licenses/LICENSE-2.0
|
||||
*
|
||||
* Unless required by applicable law or agreed to in writing, software
|
||||
* distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
|
||||
* WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
|
||||
* License for the specific language governing permissions and limitations
|
||||
* under the License.
|
||||
*/
|
||||
package io.netty.buffer.api.examples.bytetomessagedecoder;
|
||||
|
||||
import io.netty.buffer.api.Buffer;
|
||||
import io.netty.buffer.api.BufferAllocator;
|
||||
import io.netty.buffer.api.Send;
|
||||
import io.netty.channel.ChannelHandlerAdapter;
|
||||
import io.netty.channel.ChannelHandlerContext;
|
||||
|
||||
import java.util.Objects;
|
||||
|
||||
public abstract class AlternativeMessageDecoder extends ChannelHandlerAdapter {
|
||||
public static final int DEFAULT_CHUNK_SIZE = 1 << 13; // 8 KiB
|
||||
private Buffer collector;
|
||||
private BufferAllocator allocator;
|
||||
|
||||
protected AlternativeMessageDecoder() {
|
||||
allocator = initAllocator();
|
||||
collector = initCollector(allocator, DEFAULT_CHUNK_SIZE);
|
||||
}
|
||||
|
||||
protected BufferAllocator initAllocator() {
|
||||
return BufferAllocator.heap();
|
||||
}
|
||||
|
||||
protected Buffer initCollector(BufferAllocator allocator, int defaultChunkSize) {
|
||||
return allocator.allocate(defaultChunkSize);
|
||||
}
|
||||
|
||||
@Override
|
||||
public void handlerRemoved(ChannelHandlerContext ctx) throws Exception {
|
||||
drainCollector(ctx);
|
||||
collector.close();
|
||||
super.handlerRemoved(ctx);
|
||||
}
|
||||
|
||||
private void drainCollector(ChannelHandlerContext ctx) {
|
||||
boolean madeProgress;
|
||||
do {
|
||||
madeProgress = decodeAndFireRead(ctx, collector);
|
||||
} while (madeProgress);
|
||||
}
|
||||
|
||||
protected abstract boolean decodeAndFireRead(ChannelHandlerContext ctx, Buffer input);
|
||||
|
||||
public BufferAllocator getAllocator() {
|
||||
return allocator;
|
||||
}
|
||||
|
||||
public void setAllocator(BufferAllocator allocator) {
|
||||
this.allocator = Objects.requireNonNull(allocator, "BufferAllocator cannot be null.");
|
||||
}
|
||||
|
||||
@Override
|
||||
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
|
||||
if (msg instanceof Buffer) {
|
||||
try (Buffer input = (Buffer) msg) {
|
||||
processRead(ctx, input);
|
||||
}
|
||||
} else if (msg instanceof Send && ((Send<?>) msg).isInstanceOf(Buffer.class)) {
|
||||
//noinspection unchecked
|
||||
try (Buffer input = ((Send<Buffer>) msg).receive()) {
|
||||
processRead(ctx, input);
|
||||
}
|
||||
} else {
|
||||
super.channelRead(ctx, msg);
|
||||
}
|
||||
}
|
||||
|
||||
private void processRead(ChannelHandlerContext ctx, Buffer input) {
|
||||
if (collector.isOwned() && Buffer.isComposite(collector) && input.isOwned()
|
||||
&& (collector.writableBytes() == 0 || input.writerOffset() == 0)
|
||||
&& (collector.readableBytes() == 0 || input.readerOffset() == 0)
|
||||
&& collector.order() == input.order()) {
|
||||
Buffer.extendComposite(collector, input);
|
||||
drainCollector(ctx);
|
||||
return;
|
||||
}
|
||||
if (collector.isOwned()) {
|
||||
collector.ensureWritable(input.readableBytes(), DEFAULT_CHUNK_SIZE, true);
|
||||
} else {
|
||||
try (Buffer prev = collector) {
|
||||
int requiredCapacity = input.capacity() + prev.readableBytes();
|
||||
collector = allocator.allocate(Math.max(requiredCapacity, DEFAULT_CHUNK_SIZE), input.order());
|
||||
prev.copyInto(prev.readerOffset(), collector, 0, prev.readableBytes());
|
||||
collector.readerOffset(prev.readableBytes());
|
||||
}
|
||||
}
|
||||
input.copyInto(input.readerOffset(), collector, collector.writerOffset(), input.readableBytes());
|
||||
collector.writerOffset(collector.writerOffset() + input.readableBytes());
|
||||
drainCollector(ctx);
|
||||
}
|
||||
|
||||
@Override
|
||||
public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
|
||||
if (ctx.channel().config().isAutoRead()) {
|
||||
ctx.read();
|
||||
}
|
||||
ctx.fireChannelReadComplete();
|
||||
}
|
||||
}
|
||||
@ -0,0 +1,99 @@
|
||||
/*
|
||||
* Copyright 2021 The Netty Project
|
||||
*
|
||||
* The Netty Project licenses this file to you under the Apache License,
|
||||
* version 2.0 (the "License"); you may not use this file except in compliance
|
||||
* with the License. You may obtain a copy of the License at:
|
||||
*
|
||||
* https://www.apache.org/licenses/LICENSE-2.0
|
||||
*
|
||||
* Unless required by applicable law or agreed to in writing, software
|
||||
* distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
|
||||
* WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
|
||||
* License for the specific language governing permissions and limitations
|
||||
* under the License.
|
||||
*/
|
||||
package io.netty.buffer.api.examples.bytetomessagedecoder;
|
||||
|
||||
import io.netty.buffer.api.Buffer;
|
||||
import io.netty.buffer.api.BufferAllocator;
|
||||
import io.netty.channel.ChannelHandlerContext;
|
||||
import io.netty.channel.embedded.EmbeddedChannel;
|
||||
import org.junit.jupiter.api.Test;
|
||||
|
||||
import java.util.ArrayList;
|
||||
import java.util.Iterator;
|
||||
import java.util.List;
|
||||
import java.util.SplittableRandom;
|
||||
|
||||
import static io.netty.buffer.api.BufferTestSupport.toByteArray;
|
||||
import static org.assertj.core.api.Assertions.assertThat;
|
||||
import static org.junit.jupiter.api.Assertions.assertFalse;
|
||||
import static org.junit.jupiter.api.Assertions.assertTrue;
|
||||
|
||||
class AlternativeMessageDecoderTest {
|
||||
@Test
|
||||
public void splitAndParseMessagesDownThePipeline() {
|
||||
EmbeddedChannel channel = new EmbeddedChannel(new AlternativeMessageDecoder() {
|
||||
@Override
|
||||
protected boolean decodeAndFireRead(ChannelHandlerContext ctx, Buffer input) {
|
||||
// Can we read our length header?
|
||||
if (input.readableBytes() < 4) {
|
||||
return false;
|
||||
}
|
||||
|
||||
int start = input.readerOffset();
|
||||
int length = input.readInt();
|
||||
// Can we read the rest of the message?
|
||||
if (input.readableBytes() < length) {
|
||||
input.readerOffset(start);
|
||||
return false;
|
||||
}
|
||||
|
||||
// We can read our message in full.
|
||||
Buffer messageBuffer = input.bifurcate(input.readerOffset() + length);
|
||||
ctx.fireChannelRead(messageBuffer);
|
||||
return true;
|
||||
}
|
||||
});
|
||||
|
||||
List<byte[]> messages = new ArrayList<>();
|
||||
Buffer messagesBuffer = BufferAllocator.heap().allocate(132 * 1024);
|
||||
SplittableRandom rng = new SplittableRandom(42);
|
||||
for (int i = 0; i < 1000; i++) {
|
||||
byte[] message = new byte[rng.nextInt(4, 256)];
|
||||
rng.nextBytes(message);
|
||||
message[0] = (byte) (i >> 24);
|
||||
message[1] = (byte) (i >> 16);
|
||||
message[2] = (byte) (i >> 8);
|
||||
message[3] = (byte) i;
|
||||
messages.add(message);
|
||||
messagesBuffer.ensureWritable(4 + message.length, 1024, false);
|
||||
messagesBuffer.writeInt(message.length);
|
||||
for (byte b : message) {
|
||||
messagesBuffer.writeByte(b);
|
||||
}
|
||||
}
|
||||
|
||||
while (messagesBuffer.readableBytes() > 0) {
|
||||
int length = rng.nextInt(1, Math.min(500, messagesBuffer.readableBytes() + 1));
|
||||
if (length == messagesBuffer.readableBytes()) {
|
||||
channel.writeInbound(messagesBuffer);
|
||||
} else {
|
||||
channel.writeInbound(messagesBuffer.bifurcate(length));
|
||||
}
|
||||
}
|
||||
|
||||
Iterator<byte[]> expectedItr = messages.iterator();
|
||||
Buffer actualMessage;
|
||||
while ((actualMessage = channel.readInbound()) != null) {
|
||||
try (Buffer ignore = actualMessage) {
|
||||
assertTrue(expectedItr.hasNext());
|
||||
try (Buffer actual = actualMessage.slice()) {
|
||||
assertThat(toByteArray(actual)).containsExactly(expectedItr.next());
|
||||
}
|
||||
}
|
||||
}
|
||||
assertFalse(expectedItr.hasNext());
|
||||
}
|
||||
}
|
||||
Loading…
Reference in New Issue
Block a user