Inject asynchronous, synthetic messages into inbound channel handler pipeline

Question

I have a netty client reading messages from a TCP connection, and I want to add synthetic messages into the pipeline, to be processed by handlers down the line just as the normal messages.

The messages are to be injected once a second, using a periodic timer, i.e. these messages are asynchronous to the incoming messages, so I would have to invoke the pipeline somehow manually, without waiting for tcp traffic, but it will certainly have to be synchronized so the pipeline handlers are not being called concurrently.

In below simple example pipeline, the messages would have to be inserted after 1. (because otherwise the frame decoder would get confused), but before 2. (because that is supposed to process the synthetic messages, in addition to the regular ones).

How can this be done?

Answer 1

You can do this by acquiring a reference to the ChannelHandlerContext of the handler just upstream of where you want to inject the synthetic message. When you want to insert the message, call the context's fireChannelRead(Object) method, which invokes the channelRead method of the next inbound handler in the pipeline.

Netty will take care of making sure the message is processed in the correct thread. (If you're already in the channel's event loop thread, the next handler is invoked immediately; otherwise the invocation is scheduled to occur on the channel's event loop thread at the next available opportunity).

In your case, you want the context for the frame-decoder handler. ChannelPipeline has a few overloaded "context" methods that return the context for a handler. In this example, let's assume you've given you handler the name "frame-decoder", and we'll look up the context using this name. (Alternatively, you could look it up by passing a reference to the handler object itself, or by passing the Class of the handler.)

public static void injectMessage(ChannelPipeline pipeline,
                                 Object message) {
    ChannelHandlerContext ctx = pipeline.context("frame-decoder");
    ctx.fireChannelRead(message);
}