I am reading variable length messages where each message is an int + msg bytes. Messages could be any size.
Right now my strategy is to have a direct buffer that reads from the socket, which then transfers the bytes read into an array-backed nondirect buffer. However my solution has a lot of copying, some loops and just feels a bit "icky".
Is there a better solution?
Here is the read completion handler. I'm hoping it should be pretty obvious what is going on:
class DelimitedReadCompletionHandler implements CompletionHandler<Integer, Object> {
private final AsynchronousByteChannel channel;
private final Consumer<byte[]> onMessageCallback;
private final ByteBuffer channelReadBuffer; // this is a direct buffer.
private volatile ByteBuffer currentMessageBuffer;
public DelimitedReadCompletionHandler(AsynchronousByteChannel channel,
Consumer<byte[]> onMessageCallback,
ByteBuffer channelReadBuffer) {
this.channel = channel;
this.onMessageCallback = onMessageCallback;
this.channelReadBuffer = channelReadBuffer;
}
@Override
public void completed(Integer result,
Object attachment) {
channelReadBuffer.flip();
do {
if (currentMessageBuffer == null) {
if (channelReadBuffer.remaining() < Integer.BYTES) {
channelReadBuffer.compact();
break;
}
currentMessageBuffer = ByteBuffer.allocate(channelReadBuffer.getInt());
}
int copyLength = Integer.min(currentMessageBuffer.remaining(), channelReadBuffer.remaining());
for (int i = 0; i < copyLength; i++) {
currentMessageBuffer.put(channelReadBuffer.get());
}
if (!currentMessageBuffer.hasRemaining()) {
byte[] msg = currentMessageBuffer.array();
ForkJoinPool.commonPool().execute(() -> onMessageCallback.accept(msg));
currentMessageBuffer = null;
}
if (channelReadBuffer.hasRemaining()) {
continue;
}
channelReadBuffer.clear();
break;
} while (true);
channel.read(channelReadBuffer, null, this);
}
@Override
public void failed(Throwable exc,
Object attachment) {
exc.printStackTrace();
}
}