mirror of
https://github.com/apache/rocketmq.git
synced 2026-09-01 15:47:01 +08:00
* Use CompositeByteBuf to prevent mem_copy. * Fix code * Add tests * Remove useless UTs * Remove unused imports. --------- Co-authored-by: RongtongJin <jinrongtong16@mails.ucas.ac.cn>
This commit is contained in:
@@ -18,6 +18,9 @@
|
||||
package org.apache.rocketmq.remoting.netty;
|
||||
|
||||
import io.netty.buffer.ByteBuf;
|
||||
import io.netty.buffer.ByteBufAllocator;
|
||||
import io.netty.buffer.CompositeByteBuf;
|
||||
import io.netty.buffer.Unpooled;
|
||||
import io.netty.channel.ChannelHandlerContext;
|
||||
import io.netty.channel.FileRegion;
|
||||
import io.netty.handler.codec.MessageToByteEncoder;
|
||||
@@ -51,9 +54,12 @@ public class FileRegionEncoder extends MessageToByteEncoder<FileRegion> {
|
||||
WritableByteChannel writableByteChannel = new WritableByteChannel() {
|
||||
@Override
|
||||
public int write(ByteBuffer src) {
|
||||
int prev = out.writerIndex();
|
||||
out.writeBytes(src);
|
||||
return out.writerIndex() - prev;
|
||||
// To prevent mem_copy.
|
||||
CompositeByteBuf b = (CompositeByteBuf) out;
|
||||
// Have to increase writerIndex manually.
|
||||
ByteBuf unpooled = Unpooled.wrappedBuffer(src);
|
||||
b.addComponent(true, unpooled);
|
||||
return unpooled.readableBytes();
|
||||
}
|
||||
|
||||
@Override
|
||||
@@ -76,4 +82,10 @@ public class FileRegionEncoder extends MessageToByteEncoder<FileRegion> {
|
||||
msg.transferTo(writableByteChannel, transferred);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@Override
|
||||
protected ByteBuf allocateBuffer(ChannelHandlerContext ctx, FileRegion msg, boolean preferDirect) throws Exception {
|
||||
ByteBufAllocator allocator = ctx.alloc();
|
||||
return preferDirect ? allocator.compositeDirectBuffer() : allocator.compositeHeapBuffer();
|
||||
}
|
||||
}
|
||||
+3
-2
@@ -21,14 +21,15 @@ import io.netty.buffer.ByteBuf;
|
||||
import io.netty.channel.DefaultFileRegion;
|
||||
import io.netty.channel.FileRegion;
|
||||
import io.netty.channel.embedded.EmbeddedChannel;
|
||||
import org.junit.Assert;
|
||||
import org.junit.Test;
|
||||
|
||||
import java.io.BufferedOutputStream;
|
||||
import java.io.File;
|
||||
import java.io.FileOutputStream;
|
||||
import java.io.IOException;
|
||||
import java.util.Random;
|
||||
import java.util.UUID;
|
||||
import org.junit.Assert;
|
||||
import org.junit.Test;
|
||||
|
||||
public class FileRegionEncoderTest {
|
||||
|
||||
|
||||
Reference in New Issue
Block a user