andreacavalli/netty5
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

Allow to obtain RecvByteBufAllocator.Handle to allow more flexible implementations

Browse Source 
Motivation:

At the moment it's only possible for a user to set the RecvByteBufAllocator for a Channel but not access the Handle once it is assigned. This makes it hard to write more flexible implementations.

Modifications:

Add a new method to the Channel.Unsafe to allow access the the used Handle for the Channel. The RecvByteBufAllocator.Handle is created lazily.

Result:

It's possible to write more flexible implementatons that allow to adjust stuff on the fly for a Handle that is used by a Channel
This commit is contained in:
Norman Maurer 2014-08-06 14:38:23 +02:00
parent 7764b0e3de
commit c5b6808645
10 changed files with 24 additions and 42 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
transport-native-epoll/src/main/java/io/netty/channel/epoll/EpollDatagramChannel.java
View File

@ -386,7 +386,6 @@ public final class EpollDatagramChannel extends AbstractEpollChannel implements
}
final class EpollDatagramChannelUnsafe extends AbstractEpollUnsafe {
private RecvByteBufAllocator.Handle allocHandle;
@Override
public void connect(SocketAddress remote, SocketAddress local, ChannelPromise channelPromise) {
@ -419,10 +418,7 @@ public final class EpollDatagramChannel extends AbstractEpollChannel implements
@Override
void epollInReady() {
DatagramChannelConfig config = config();
RecvByteBufAllocator.Handle allocHandle = this.allocHandle;
if (allocHandle == null) {
this.allocHandle = allocHandle = config.getRecvByteBufAllocator().newHandle();
}
RecvByteBufAllocator.Handle allocHandle = unsafe().recvBufAllocHandle();
assert eventLoop().inEventLoop();
final ChannelPipeline pipeline = pipeline();

5
transport-native-epoll/src/main/java/io/netty/channel/epoll/EpollSocketChannel.java
View File

@ -693,10 +693,7 @@ public final class EpollSocketChannel extends AbstractEpollChannel implements So
final ChannelConfig config = config();
final ChannelPipeline pipeline = pipeline();
final ByteBufAllocator allocator = config.getAllocator();
RecvByteBufAllocator.Handle allocHandle = this.allocHandle;
if (allocHandle == null) {
this.allocHandle = allocHandle = config.getRecvByteBufAllocator().newHandle();
}
RecvByteBufAllocator.Handle allocHandle = unsafe().recvBufAllocHandle();
ByteBuf byteBuf = null;
boolean close = false;

7
transport-sctp/src/main/java/io/netty/channel/sctp/nio/NioSctpChannel.java
View File

@ -68,8 +68,6 @@ public class NioSctpChannel extends AbstractNioMessageChannel implements io.nett
private final NotificationHandler<?> notificationHandler;
private RecvByteBufAllocator.Handle allocHandle;
private static SctpChannel newSctpChannel() {
try {
return SctpChannel.open();
@ -265,10 +263,7 @@ public class NioSctpChannel extends AbstractNioMessageChannel implements io.nett
protected int doReadMessages(List<Object> buf) throws Exception {
SctpChannel ch = javaChannel();
RecvByteBufAllocator.Handle allocHandle = this.allocHandle;
if (allocHandle == null) {
this.allocHandle = allocHandle = config().getRecvByteBufAllocator().newHandle();
}
RecvByteBufAllocator.Handle allocHandle = unsafe().recvBufAllocHandle();
ByteBuf buffer = allocHandle.allocate(config().getAllocator());
boolean free = true;
try {

7
transport-sctp/src/main/java/io/netty/channel/sctp/oio/OioSctpChannel.java
View File

@ -76,8 +76,6 @@ public class OioSctpChannel extends AbstractOioMessageChannel
private final NotificationHandler<?> notificationHandler;
private RecvByteBufAllocator.Handle allocHandle;
private static SctpChannel openChannel() {
try {
return SctpChannel.open();
@ -187,10 +185,7 @@ public class OioSctpChannel extends AbstractOioMessageChannel
Set<SelectionKey> reableKeys = readSelector.selectedKeys();
try {
for (SelectionKey ignored : reableKeys) {
RecvByteBufAllocator.Handle allocHandle = this.allocHandle;
if (allocHandle == null) {
this.allocHandle = allocHandle = config().getRecvByteBufAllocator().newHandle();
}
RecvByteBufAllocator.Handle allocHandle = unsafe().recvBufAllocHandle();
ByteBuf buffer = allocHandle.allocate(config().getAllocator());
boolean free = true;

9
transport/src/main/java/io/netty/channel/AbstractChannel.java
View File

@ -419,8 +419,17 @@ public abstract class AbstractChannel extends DefaultAttributeMap implements Cha
protected abstract class AbstractUnsafe implements Unsafe {
private ChannelOutboundBuffer outboundBuffer = new ChannelOutboundBuffer(AbstractChannel.this);
private RecvByteBufAllocator.Handle recvHandle;
private boolean inFlush0;
@Override
public RecvByteBufAllocator.Handle recvBufAllocHandle() {
if (recvHandle == null) {
recvHandle = config().getRecvByteBufAllocator().newHandle();
}
return recvHandle;
}
@Override
public final ChannelHandlerInvoker invoker() {
// return the unwrapped invoker.

6
transport/src/main/java/io/netty/channel/Channel.java
View File

@ -464,6 +464,12 @@ public interface Channel extends AttributeMap, Comparable<Channel> {
*/
interface Unsafe {
/**
* Return the assigned {@link RecvByteBufAllocator.Handle} which will be used to allocate {@link ByteBuf}'s when
* receiving data.
*/
RecvByteBufAllocator.Handle recvBufAllocHandle();
/**
* Returns the {@link ChannelHandlerInvoker} which is used by default unless specified by a user.
*/

6
transport/src/main/java/io/netty/channel/nio/AbstractNioByteChannel.java
View File

@ -58,7 +58,6 @@ public abstract class AbstractNioByteChannel extends AbstractNioChannel {
}
private final class NioByteUnsafe extends AbstractNioUnsafe {
private RecvByteBufAllocator.Handle allocHandle;
private void closeOnRead(ChannelPipeline pipeline) {
SelectionKey key = selectionKey();
@ -102,10 +101,7 @@ public abstract class AbstractNioByteChannel extends AbstractNioChannel {
final ChannelPipeline pipeline = pipeline();
final ByteBufAllocator allocator = config.getAllocator();
final int maxMessagesPerRead = config.getMaxMessagesPerRead();
RecvByteBufAllocator.Handle allocHandle = this.allocHandle;
if (allocHandle == null) {
this.allocHandle = allocHandle = config.getRecvByteBufAllocator().newHandle();
}
RecvByteBufAllocator.Handle allocHandle = recvBufAllocHandle();
ByteBuf byteBuf = null;
int messages = 0;

6
transport/src/main/java/io/netty/channel/oio/AbstractOioByteChannel.java
View File

@ -39,7 +39,6 @@ public abstract class AbstractOioByteChannel extends AbstractOioChannel {
" (expected: " + StringUtil.simpleClassName(ByteBuf.class) + ", " +
StringUtil.simpleClassName(FileRegion.class) + ')';
private RecvByteBufAllocator.Handle allocHandle;
private volatile boolean inputShutdown;
/**
@ -82,10 +81,7 @@ public abstract class AbstractOioByteChannel extends AbstractOioChannel {
final ChannelConfig config = config();
final ChannelPipeline pipeline = pipeline();
RecvByteBufAllocator.Handle allocHandle = this.allocHandle;
if (allocHandle == null) {
this.allocHandle = allocHandle = config.getRecvByteBufAllocator().newHandle();
}
RecvByteBufAllocator.Handle allocHandle = unsafe().recvBufAllocHandle();
ByteBuf byteBuf = allocHandle.allocate(alloc());

7
transport/src/main/java/io/netty/channel/socket/nio/NioDatagramChannel.java
View File

@ -72,7 +72,6 @@ public final class NioDatagramChannel
private final DatagramChannelConfig config;
private Map<InetAddress, List<MembershipKey>> memberships;
private RecvByteBufAllocator.Handle allocHandle;
private static DatagramChannel newSocket(SelectorProvider provider) {
try {
@ -230,10 +229,8 @@ public final class NioDatagramChannel
protected int doReadMessages(List<Object> buf) throws Exception {
DatagramChannel ch = javaChannel();
DatagramChannelConfig config = config();
RecvByteBufAllocator.Handle allocHandle = this.allocHandle;
if (allocHandle == null) {
this.allocHandle = allocHandle = config.getRecvByteBufAllocator().newHandle();
}
RecvByteBufAllocator.Handle allocHandle = unsafe().recvBufAllocHandle();
ByteBuf data = allocHandle.allocate(config.getAllocator());
boolean free = true;
try {

7
transport/src/main/java/io/netty/channel/socket/oio/OioDatagramChannel.java
View File

@ -71,8 +71,6 @@ public class OioDatagramChannel extends AbstractOioMessageChannel
private final DatagramChannelConfig config;
private final java.net.DatagramPacket tmpPacket = new java.net.DatagramPacket(EmptyArrays.EMPTY_BYTES, 0);
private RecvByteBufAllocator.Handle allocHandle;
private static MulticastSocket newSocket() {
try {
return new MulticastSocket(null);
@ -202,10 +200,7 @@ public class OioDatagramChannel extends AbstractOioMessageChannel
@Override
protected int doReadMessages(List<Object> buf) throws Exception {
DatagramChannelConfig config = config();
RecvByteBufAllocator.Handle allocHandle = this.allocHandle;
if (allocHandle == null) {
this.allocHandle = allocHandle = config.getRecvByteBufAllocator().newHandle();
}
RecvByteBufAllocator.Handle allocHandle = unsafe().recvBufAllocHandle();
ByteBuf data = config.getAllocator().heapBuffer(allocHandle.guess());
boolean free = true;