org.apache.reef.wake.remote.transport.netty

Class ChunkedReadWriteHandler

java.lang.Object

io.netty.channel.ChannelHandlerAdapter

io.netty.channel.ChannelInboundHandlerAdapter

io.netty.channel.ChannelDuplexHandler

io.netty.handler.stream.ChunkedWriteHandler

org.apache.reef.wake.remote.transport.netty.ChunkedReadWriteHandler

All Implemented Interfaces:

io.netty.channel.ChannelHandler, io.netty.channel.ChannelInboundHandler, io.netty.channel.ChannelOutboundHandler

public class ChunkedReadWriteHandler
extends io.netty.handler.stream.ChunkedWriteHandler

Thin wrapper around ChunkedWriteHandler

ChunkedWriteHandler only handles the down stream parts and just emits the chunks up stream. So we add an upstream handler that aggregates the chunks into its original form. This is guaranteed to be thread serial so state can be shared.

On the down stream side, we just decorate the original message with its size and allow the thread-serial base class to actually handle the chunking. We need to be careful since the decoration itself has to be thread-safe since netty does not guarantee thread serial access to down stream handlers.

We do not need to tag the writes since the base class ChunkedWriteHandler serializes access to the channel and first write will complete before the second begins.

Nested Class Summary

Nested classes/interfaces inherited from interface io.netty.channel.ChannelHandler

io.netty.channel.ChannelHandler.Sharable

Field Summary

Fields

Modifier and Type	Field and Description
static int	INT_SIZE

Constructor Summary

Constructors

Constructor and Description
ChunkedReadWriteHandler()

Method Summary

Methods

Modifier and Type	Method and Description
void	channelRead(io.netty.channel.ChannelHandlerContext ctx, Object msg)
void	write(io.netty.channel.ChannelHandlerContext ctx, Object msg, io.netty.channel.ChannelPromise promise) Thread-safe since there is no shared instance state.

Methods inherited from class io.netty.handler.stream.ChunkedWriteHandler

channelInactive, channelWritabilityChanged, flush, handlerAdded, resumeTransfer

Methods inherited from class io.netty.channel.ChannelDuplexHandler

bind, close, connect, deregister, disconnect, read

Methods inherited from class io.netty.channel.ChannelInboundHandlerAdapter

channelActive, channelReadComplete, channelRegistered, channelUnregistered, exceptionCaught, userEventTriggered

Methods inherited from class io.netty.channel.ChannelHandlerAdapter

handlerRemoved, isSharable

Methods inherited from class java.lang.Object

equals, getClass, hashCode, notify, notifyAll, toString, wait, wait, wait

Methods inherited from interface io.netty.channel.ChannelHandler

exceptionCaught, handlerRemoved

Field Detail

INT_SIZE

public static final int INT_SIZE

See Also:

Constant Field Values

Constructor Detail

ChunkedReadWriteHandler

public ChunkedReadWriteHandler()

Method Detail

channelRead

public void channelRead(io.netty.channel.ChannelHandlerContext ctx,
               Object msg)
                 throws Exception

Specified by:

channelRead in interface io.netty.channel.ChannelInboundHandler

Overrides:

channelRead in class io.netty.channel.ChannelInboundHandlerAdapter

Throws:

Exception

See Also:

org.jboss.netty.handler.stream.ChunkedWriteHandler#handleUpstream(org.jboss.netty.channel.ChannelHandlerContext, org.jboss.netty.channel.ChannelEvent)

write

public void write(io.netty.channel.ChannelHandlerContext ctx,
         Object msg,
         io.netty.channel.ChannelPromise promise)
           throws Exception

Thread-safe since there is no shared instance state. Just prepend size to the message and stream it through a chunked stream and let the base method handle the actual chunking.

We do not need to tag the writes since the base class ChunkedWriteHandler serializes access to the channel and first write will complete before the second begins.

Specified by:

write in interface io.netty.channel.ChannelOutboundHandler

Overrides:

write in class io.netty.handler.stream.ChunkedWriteHandler

Throws:

Exception