05 Service compilation layer Pipeline, how various Handlers coordinate #

In the previous lesson, we learned that the EventLoop can be considered as the scheduling center of Netty, responsible for listening to various types of events, such as I/O events, signal events, and timer events. However, the actual business logic is implemented by the ChannelHandlers defined in the ChannelPipeline. ChannelPipeline and ChannelHandler are the most frequently used components in application development using Netty. The core components of the Netty service orchestration layer, ChannelPipeline and ChannelHandler, provide users with complete control over I/O events. In today’s lesson, we will delve into how Netty utilizes these two components to manipulate data.

Before we dive into this lesson, let me raise a few questions.

• What is the relationship between ChannelPipeline and ChannelHandler? How do they work together?
• What types of ChannelHandlers are there, and what are the differences?
• How are I/O events propagated in Netty?

I hope after completing this lesson, you will find answers to these questions.

Overview of ChannelPipeline #

The literal meaning of the word “Pipeline” is a pipeline or assembly line. In Netty, it plays a similar role to that of a factory assembly line. The original network byte stream passes through the Pipeline, where it is progressively processed and packaged until it is transformed into the final product. After the initial learning of the core components in the previous lessons, we already have a preliminary impression of ChannelPipeline: it is the core processing chain of Netty, used for dynamic orchestration and ordered propagation of network events.

Today, we will explore the implementation principles of ChannelPipeline from the following aspects:

• Internal structure of ChannelPipeline
• ChannelHandler interface design
• Event propagation mechanism in ChannelPipeline
• Exception propagation mechanism in ChannelPipeline

Internal Structure of ChannelPipeline #

Firstly, we need to clarify what the internal structure of ChannelPipeline looks like in order to understand the processing flow of ChannelPipeline. ChannelPipeline, as the core orchestration component of Netty, is responsible for scheduling various types of ChannelHandlers, and the actual processing of data is completed by ChannelHandlers.

ChannelPipeline can be seen as a container for ChannelHandlers. It consists of a set of ChannelHandler instances, which are linked together through a bidirectional linked list as shown in the following diagram. When an I/O read/write event is triggered, ChannelPipeline will sequentially invoke the ChannelHandler list to intercept and process the data of the Channel.

From the above diagram, we can see that each Channel is bound to a ChannelPipeline, and each ChannelPipeline contains multiple ChannelHandlerContexts. All the ChannelHandlerContexts are connected by a bidirectional linked list. Also, because each ChannelHandler corresponds to a ChannelHandlerContext, in fact, ChannelPipeline maintains the relationship between ChannelHandlerContext and itself. You may wonder why there is an additional layer of ChannelHandlerContext encapsulation here? Actually, this is a commonly used programming concept. ChannelHandlerContext is used to save the context of a ChannelHandler. ChannelHandlerContext contains all the lifecycle events of a ChannelHandler, such as connect, bind, read, flush, write, close, etc. Imagine that if there is no ChannelHandlerContext encapsulation, then we would have to implement the common logic before and after in each ChannelHandler when passing between ChannelHandlers. Although this can solve the problem, the code structure coupling would be very inelegant.

According to the flow direction of network data, ChannelPipeline is divided into inbound ChannelInboundHandler and outbound ChannelOutboundHandler. In the process of communication between the client and the server, the data sent from the client to the server is called outbound, and vice versa is called inbound. The data is processed by a series of InboundHandlers and then inbound. After being processed by the opposite direction’s OutboundHandlers, it becomes outbound. This is illustrated in the following figure. The decoder we often use is an inbound operation, and the encoder is an outbound operation. When the server receives client data, it needs to go through the inbound processing of the Decoder first and then inform the client through the outbound processing of the Encoder.

Next, let’s analyze the construction of the bidirectional linked list of ChannelPipeline in detail. The bidirectional linked list of ChannelPipeline maintains the head node and the tail node of HeadContext and TailContext respectively. The ChannelHandlers we customize will be inserted between the Head and Tail nodes, which have been implemented by default in Netty and play a crucial role in the ChannelPipeline. First, let’s look at the inheritance relationship between HeadContext and TailContext, as shown in the following figure.

HeadContext is both an inbound and outbound processor. It implements both ChannelInboundHandler and ChannelOutboundHandler. The entry point for writing network data is completed by the HeadContext node. As the head node of the Pipeline, HeadContext is responsible for reading data and starting to propagate Inbound events. After the data is processed, it will be transmitted in the opposite direction through the Outbound processors and eventually passed to HeadContext. Therefore, HeadContext is also the last stop for processing Outbound events. In addition, before propagating events, HeadContext will also perform some pre-processing.

TailContext only implements the ChannelInboundHandler interface. It is executed in the last step of the ChannelInboundHandler call chain and is mainly used to terminate the propagation of Inbound events, such as releasing Message data resources, etc. The TailContext node is the first stop for the propagation of Outbound events, it only passes Outbound events to the previous node.

From the perspective of the entire ChannelPipeline call chain, if an event propagation is triggered directly by the Channel, the call chain will span the entire ChannelPipeline. However, the same method can also be triggered by one of the ChannelHandlerContext, in which case the event propagation will start from the current ChannelHandler, and the process will not span from the head to the tail. In certain scenarios, this can improve program performance.

ChannelHandler Interface Design #

Before learning about the ChannelPipeline event propagation mechanism, we need to understand the lifecycle of I/O events. The entire ChannelHandler is designed around the lifecycle of I/O events, such as establishing connections, reading data, writing data, destroying connections, etc. ChannelHandler has two important subinterfaces: ChannelInboundHandler and ChannelOutboundHandler, which intercept various types of I/O events inbound and outbound respectively.

1. Event callback methods of ChannelInboundHandler and trigger timing.

2. Event callback methods of ChannelOutboundHandler and trigger timing.

The event callback methods of ChannelOutboundHandler are very clear. You can see each operation’s corresponding callback method directly through the ChannelOutboundHandler interface list, as shown in the figure below. Each callback method is triggered before the corresponding operation executes and no further explanation will be given here. In addition, most of the interfaces in ChannelOutboundHandler take a ChannelPromise parameter, so that notifications can be received promptly when the operation is completed.

Event Propagation Mechanism #

In the previous section, we introduced that the ChannelPipeline can be divided into two types of handlers: ChannelInboundHandler for inbound events and ChannelOutboundHandler for outbound events.

Let’s experience the event propagation mechanism of ChannelPipeline together through a code example.

serverBootstrap.childHandler(new ChannelInitializer<SocketChannel>() {

  @Override
  public void initChannel(SocketChannel ch) {

    ch.pipeline()
      .addLast(new SampleInBoundHandler("SampleInBoundHandlerA", false))
      .addLast(new SampleInBoundHandler("SampleInBoundHandlerB", false))
      .addLast(new SampleInBoundHandler("SampleInBoundHandlerC", true));

    ch.pipeline()
      .addLast(new SampleOutBoundHandler("SampleOutBoundHandlerA"))
      .addLast(new SampleOutBoundHandler("SampleOutBoundHandlerB"))
      .addLast(new SampleOutBoundHandler("SampleOutBoundHandlerC"));
  }

});

public class SampleInBoundHandler extends ChannelInboundHandlerAdapter {

  private final String name;
  private final boolean flush;

  public SampleInBoundHandler(String name, boolean flush) {
    this.name = name;
    this.flush = flush;
  }

  @Override
  public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
    System.out.println("InBoundHandler: " + name);

    if (flush) {
      ctx.channel().writeAndFlush(msg);
    } else {
      super.channelRead(ctx, msg);
    }
  }

}

public class SampleOutBoundHandler extends ChannelOutboundHandlerAdapter {

  private final String name;

  public SampleOutBoundHandler(String name) {
    this.name = name;
  }
}

In the code example above, we can see how the inbound and outbound handlers are added to the ChannelPipeline. The inbound handlers are SampleInBoundHandlerA, SampleInBoundHandlerB, and SampleInBoundHandlerC, while the outbound handlers are SampleOutBoundHandlerA, SampleOutBoundHandlerB, and SampleOutBoundHandlerC. }

```java
@Override
public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception {
    System.out.println("OutBoundHandler: " + name);
    super.write(ctx, msg, promise);
}
}

By using the addLast method of the Pipeline, three InboundHandlers and OutboundHandlers are added, and the order of addition is A -> B -> C. The internal structure of the ChannelPipeline after initialization can be represented by the following figure.

When the client sends a request to the server, it triggers the channelRead event of the SampleInBoundHandler chain. After being processed by the SampleInBoundHandler chain, the writeAndFlush method is called in SampleInBoundHandlerC to write back data to the client, which in turn triggers the write event of the SampleOutBoundHandler chain. Finally, let’s take a look at the console output of the code example:

It can be seen that the direction of event propagation for Inbound events and Outbound events is different. The Inbound events propagate from head to tail, while the Outbound events propagate from tail to head, which are opposite to each other. It is important to understand the order of event propagation in Netty application programming. I recommend that you simulate the scenario of the client and server when designing the system and draw the internal structure diagram of the ChannelPipeline to avoid confusion in the calling relationship.

Exception Propagation Mechanism #

The implementation of ChannelPipeline event propagation uses the classic chain of responsibility pattern, and the chain of calls is interconnected. So what happens if an exception occurs in one of the nodes of the chain? Let’s simulate a business logic exception by modifying the implementation of the SampleInBoundHandler:

public class SampleInBoundHandler extends ChannelInboundHandlerAdapter {

    private final String name;

    private final boolean flush;

    public SampleInBoundHandler(String name, boolean flush) {

        this.name = name;

        this.flush = flush;

    }

    @Override

    public void channelRead(ChannelHandlerContext ctx, Object msg) {

        System.out.println("InBoundHandler: " + name);

        if (flush) {

            ctx.channel().writeAndFlush(msg);

        } else {

            throw new RuntimeException("InBoundHandler: " + name);

        }

    }

    @Override

    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {

        System.out.println("InBoundHandlerException: " + name);

        ctx.fireExceptionCaught(cause);

    }

}

In the channelRead event handling, the first A node throws a RuntimeException. At the same time, we override the exceptionCaught method in ChannelInboundHandlerAdapter, and add a print statement at the beginning for observing the behavior of exception propagation. Let’s take a look at the console output of the code running:

From the output, it can be seen that ctx.fireExceptionCaught propagates the exception from the head node to the tail node in order. If the user does not intercept and handle the exception, it will finally be handled by the tail node. You can find the specific implementation in the source code of TailContext:

protected void onUnhandledInboundException(Throwable cause) {

    try {

        logger.warn(
                "An exceptionCaught() event was fired, and it reached at the tail of the pipeline. " +
                        "It usually means the last handler in the pipeline did not handle the exception.",
                cause);

    } finally {

        ReferenceCountUtil.release(cause);

    }

}

Although TailContext in Netty provides a fallback exception handling logic, in many scenarios, it may not meet our requirements. If you need to intercept a specific type of exception and perform corresponding exception handling, how should you implement it? Let’s move on to the next section.

Best Practices for Exception Handling #

In the process of Netty application development, a good exception handling mechanism will greatly facilitate the troubleshooting process. Therefore, it is recommended for users to intercept exceptions uniformly, and then implement a more complete exception handling mechanism based on actual business scenarios. Through the study of exception propagation mechanism, we should realize that the best way is to add a unified exception handler at the end of the ChannelPipeline. The internal structure of the ChannelPipeline at this time is shown in the following diagram.

The following code shows an example of a user-defined exception handler:

public class ExceptionHandler extends ChannelDuplexHandler {

    @Override

    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {

        if (cause instanceof RuntimeException) {

            System.out.println("Handle Business Exception Success.");

        }

    }

}

After adding the unified exception handler, it can be seen that the exception has been intercepted and handled gracefully. This is also the recommended best practice for exception handling in Netty.

Summary #

In this lesson, we have analyzed in depth the design principles and event propagation mechanism of the Pipeline. Have you found the answers to the questions I initially raised during the course? Let me briefly summarize:

• The ChannelPipeline is a bidirectional linked list structure that contains both ChannelInboundHandlers and ChannelOutboundHandlers.
• The ChannelHandlerContext encapsulates the ChannelHandler, and each ChannelHandler corresponds to a ChannelHandlerContext. In fact, the ChannelPipeline maintains the relationship with ChannelHandlerContext.
• The direction of event propagation for Inbound events and Outbound events is opposite. Inbound events propagate from head to tail, while Outbound events propagate from tail to head.
• The order of exception event handling is the same as the order in which ChannelHandlers are added, and it propagates sequentially. It has nothing to do with Inbound events and Outbound events.

The ingenious design ideas of ChannelPipeline are worth learning from and applying. I recommend that interested students can delve into the core source code of this component. In the upcoming source code section, we will continue to explore the ChannelPipeline component in depth.