io.micronaut.http.netty.stream

Class HttpStreamsServerHandler

java.lang.Object

io.netty.channel.ChannelHandlerAdapter

io.netty.channel.ChannelInboundHandlerAdapter

io.netty.channel.ChannelDuplexHandler

io.micronaut.http.netty.stream.HttpStreamsServerHandler

All Implemented Interfaces:

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

@Internal
public class HttpStreamsServerHandler
extends io.netty.channel.ChannelDuplexHandler

Handler that reads HttpRequest messages followed by HttpContent messages and produces StreamedHttpRequest messages, and converts written StreamedHttpResponse messages into HttpResponse messages followed by HttpContent messages.

This allows request and response bodies to be handled using reactive streams.

There are two types of messages that this handler will send down the chain, StreamedHttpRequest, and FullHttpRequest. If ChannelOption.AUTO_READ is false for the channel, then any StreamedHttpRequest messages must be subscribed to consume the body, otherwise it's possible that no read will be done of the messages.

There are three types of messages that this handler accepts for writing, StreamedHttpResponse, WebSocketHttpResponse and FullHttpResponse. Writing any other messages may potentially lead to HTTP message mangling.

As long as messages are returned in the order that they arrive, this handler implicitly supports HTTP pipelining.

Author:

jroper, Graeme Rocher

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 String	HANDLER_BODY_PUBLISHER

Constructor Summary

Constructors

Constructor and Description
HttpStreamsServerHandler() Default constructor.
HttpStreamsServerHandler(List<io.netty.channel.ChannelHandler> dependentHandlers) Create a new handler that is depended on by the given handlers.

Method Summary

Modifier and Type	Method and Description
protected void	bodyRequested(io.netty.channel.ChannelHandlerContext ctx) Invoked every time a read of the incoming body is requested by the subscriber.
void	channelRead(io.netty.channel.ChannelHandlerContext ctx, Object msg)
void	channelReadComplete(io.netty.channel.ChannelHandlerContext ctx)
void	channelWritabilityChanged(io.netty.channel.ChannelHandlerContext ctx)
protected void	consumedInMessage(io.netty.channel.ChannelHandlerContext ctx) Invoked when an incoming message is fully consumed.
protected io.netty.handler.codec.http.HttpRequest	createEmptyMessage(io.netty.handler.codec.http.HttpRequest request) Create an empty incoming message.
protected io.netty.handler.codec.http.HttpRequest	createStreamedMessage(io.netty.handler.codec.http.HttpRequest httpRequest, Publisher<? extends io.netty.handler.codec.http.HttpContent> stream) Create a streamed incoming message with the given stream.
protected int	getStreamId(Object msg) Gets the stream ID from the message.
void	handlerRemoved(io.netty.channel.ChannelHandlerContext ctx)
protected boolean	hasBody(io.netty.handler.codec.http.HttpRequest request) Whether the given incoming message has a body.
protected boolean	isClient()
protected boolean	isValidInMessage(Object msg)
protected boolean	isValidOutMessage(Object msg)
protected void	receivedInMessage(io.netty.channel.ChannelHandlerContext ctx) Invoked when an incoming message is first received.
protected void	receivedOutMessage(io.netty.channel.ChannelHandlerContext ctx) Invoked when an outgoing message is first received.
protected void	sentOutMessage(io.netty.channel.ChannelHandlerContext ctx) Invoked when an outgoing message is fully sent.
protected void	subscribeSubscriberToStream(StreamedHttpMessage msg, Subscriber<io.netty.handler.codec.http.HttpContent> subscriber) Subscribe the given subscriber to the given streamed message.
protected void	unbufferedWrite(io.netty.channel.ChannelHandlerContext ctx, io.netty.handler.codec.http.HttpResponse message, io.netty.channel.ChannelPromise promise)
void	write(io.netty.channel.ChannelHandlerContext ctx, Object msg, io.netty.channel.ChannelPromise promise)

Methods inherited from class io.netty.channel.ChannelDuplexHandler

bind, close, connect, deregister, disconnect, flush, read

Methods inherited from class io.netty.channel.ChannelInboundHandlerAdapter

channelActive, channelInactive, channelRegistered, channelUnregistered, exceptionCaught, userEventTriggered

Methods inherited from class io.netty.channel.ChannelHandlerAdapter

ensureNotSharable, handlerAdded, isSharable

Methods inherited from class java.lang.Object

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

Methods inherited from interface io.netty.channel.ChannelHandler

exceptionCaught, handlerAdded

Field Detail

HANDLER_BODY_PUBLISHER

public static final String HANDLER_BODY_PUBLISHER

See Also:

Constant Field Values

Constructor Detail

HttpStreamsServerHandler

public HttpStreamsServerHandler()

Default constructor.

HttpStreamsServerHandler

public HttpStreamsServerHandler(List<io.netty.channel.ChannelHandler> dependentHandlers)

Create a new handler that is depended on by the given handlers.

The list of dependent handlers will be removed from the chain when this handler is removed from the chain, for example, when the connection is upgraded to use websockets. This is useful, for example, for removing the reactive streams publisher/subscriber from the chain in that event.

Parameters:

dependentHandlers - The handlers that depend on this handler.

Method Detail

hasBody

protected boolean hasBody(io.netty.handler.codec.http.HttpRequest request)

Whether the given incoming message has a body.

Parameters:

request - The incoming message

Returns:

Whether the incoming message has body

createEmptyMessage

protected io.netty.handler.codec.http.HttpRequest createEmptyMessage(io.netty.handler.codec.http.HttpRequest request)

Create an empty incoming message. This must be of type FullHttpMessage, and is invoked when we've determined that an incoming message can't have a body, so we send it on as a FullHttpMessage.

Parameters:

request - The incoming message

Returns:

An empty incoming message

createStreamedMessage

protected io.netty.handler.codec.http.HttpRequest createStreamedMessage(io.netty.handler.codec.http.HttpRequest httpRequest,
                                                                        Publisher<? extends io.netty.handler.codec.http.HttpContent> stream)

Create a streamed incoming message with the given stream.

Parameters:

httpRequest - The incoming message

stream - The publisher for the Http Content

Returns:

An streamed incoming message

channelRead

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

Specified by:

channelRead in interface io.netty.channel.ChannelInboundHandler

Throws:

Exception

receivedInMessage

protected void receivedInMessage(io.netty.channel.ChannelHandlerContext ctx)

Invoked when an incoming message is first received.

Overridden by sub classes for state tracking.

Parameters:

ctx - The channel handler context

sentOutMessage

protected void sentOutMessage(io.netty.channel.ChannelHandlerContext ctx)

Invoked when an outgoing message is fully sent.

Overridden by sub classes for state tracking.

Parameters:

ctx - The channel handler context

unbufferedWrite

protected void unbufferedWrite(io.netty.channel.ChannelHandlerContext ctx,
                               io.netty.handler.codec.http.HttpResponse message,
                               io.netty.channel.ChannelPromise promise)

Parameters:

ctx - The channel handler context

message - The message

promise - The promise

isValidOutMessage

protected boolean isValidOutMessage(Object msg)

Parameters:

msg - The message

Returns:

True if the handler should write the message

consumedInMessage

protected void consumedInMessage(io.netty.channel.ChannelHandlerContext ctx)

Invoked when an incoming message is fully consumed.

Overridden by sub classes for state tracking.

Parameters:

ctx - The channel handler context

bodyRequested

protected void bodyRequested(io.netty.channel.ChannelHandlerContext ctx)

Invoked every time a read of the incoming body is requested by the subscriber.

Provided so that the server subclass can intercept this to send a 100 continue response.

Parameters:

ctx - The channel handler context

isClient

protected final boolean isClient()

Returns:

Whether this is the client stream handler.

handlerRemoved

public void handlerRemoved(io.netty.channel.ChannelHandlerContext ctx)
                    throws Exception

Specified by:

handlerRemoved in interface io.netty.channel.ChannelHandler

Overrides:

handlerRemoved in class io.netty.channel.ChannelHandlerAdapter

Throws:

Exception

receivedOutMessage

protected void receivedOutMessage(io.netty.channel.ChannelHandlerContext ctx)

Invoked when an outgoing message is first received.

Overridden by sub classes for state tracking.

Parameters:

ctx - The channel handler context

subscribeSubscriberToStream

protected void subscribeSubscriberToStream(StreamedHttpMessage msg,
                                           Subscriber<io.netty.handler.codec.http.HttpContent> subscriber)

Subscribe the given subscriber to the given streamed message.

Provided so that the client subclass can intercept this to hold off sending the body of an expect 100 continue request.

Parameters:

msg - The streamed Http message

subscriber - The subscriber for the Http Content

getStreamId

protected int getStreamId(Object msg)

Gets the stream ID from the message.

Parameters:

msg - The message

Returns:

The stream id

channelReadComplete

public void channelReadComplete(io.netty.channel.ChannelHandlerContext ctx)
                         throws Exception

Specified by:

channelReadComplete in interface io.netty.channel.ChannelInboundHandler

Overrides:

channelReadComplete in class io.netty.channel.ChannelInboundHandlerAdapter

Throws:

Exception

write

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

Specified by:

write in interface io.netty.channel.ChannelOutboundHandler

Overrides:

write in class io.netty.channel.ChannelDuplexHandler

Throws:

Exception

channelWritabilityChanged

public void channelWritabilityChanged(io.netty.channel.ChannelHandlerContext ctx)

Specified by:

channelWritabilityChanged in interface io.netty.channel.ChannelInboundHandler

Overrides:

channelWritabilityChanged in class io.netty.channel.ChannelInboundHandlerAdapter

isValidInMessage

protected boolean isValidInMessage(Object msg)

Parameters:

msg - The message

Returns:

True if the handler should read the message