查看本类的 API文档回源码主页

   /*
    * Copyright 2016 The Netty Project
    *
    * The Netty Project licenses this file to you under the Apache License, version
    * 2.0 (the "License"); you may not use this file except in compliance with the
    * License. You may obtain a copy of the License at:
    *
    * https://www.apache.org/licenses/LICENSE-2.0
    *
   * Unless required by applicable law or agreed to in writing, software
   * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
   * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
   * License for the specific language governing permissions and limitations under
   * the License.
   */
  package io.netty.handler.flow;
  
  import java.util.ArrayDeque;
  import java.util.Queue;
  
  import io.netty.channel.ChannelConfig;
  import io.netty.channel.ChannelDuplexHandler;
  import io.netty.channel.ChannelHandler;
  import io.netty.channel.ChannelHandlerContext;
  import io.netty.handler.codec.ByteToMessageDecoder;
  import io.netty.handler.codec.MessageToByteEncoder;
  import io.netty.util.ReferenceCountUtil;
  import io.netty.util.internal.ObjectPool;
  import io.netty.util.internal.ObjectPool.Handle;
  import io.netty.util.internal.ObjectPool.ObjectCreator;
  import io.netty.util.internal.logging.InternalLogger;
  import io.netty.util.internal.logging.InternalLoggerFactory;
  
  /**
   * The {@link FlowControlHandler} ensures that only one message per {@code read()} is sent downstream.
   *
   * Classes such as {@link ByteToMessageDecoder} or {@link MessageToByteEncoder} are free to emit as
   * many events as they like for any given input. A channel's auto reading configuration doesn't usually
   * apply in these scenarios. This is causing problems in downstream {@link ChannelHandler}s that would
   * like to hold subsequent events while they're processing one event. It's a common problem with the
   * {@code HttpObjectDecoder} that will very often fire an {@code HttpRequest} that is immediately followed
   * by a {@code LastHttpContent} event.
   *
   * <pre>{@code
   * ChannelPipeline pipeline = ...;
   *
   * pipeline.addLast(new HttpServerCodec());
   * pipeline.addLast(new FlowControlHandler());
   *
   * pipeline.addLast(new MyExampleHandler());
   *
   * class MyExampleHandler extends ChannelInboundHandlerAdapter {
   *   @Override
   *   public void channelRead(ChannelHandlerContext ctx, Object msg) {
   *     if (msg instanceof HttpRequest) {
   *       ctx.channel().config().setAutoRead(false);
   *
   *       // The FlowControlHandler will hold any subsequent events that
   *       // were emitted by HttpObjectDecoder until auto reading is turned
   *       // back on or Channel#read() is being called.
   *     }
   *   }
   * }
   * }</pre>
   *
   * @see ChannelConfig#setAutoRead(boolean)
   */
  public class FlowControlHandler extends ChannelDuplexHandler {
      private static final InternalLogger logger = InternalLoggerFactory.getInstance(FlowControlHandler.class);
  
      private final boolean releaseMessages;
  
      private RecyclableArrayDeque queue;
  
      private ChannelConfig config;
  
      private boolean shouldConsume;
  
      public FlowControlHandler() {
          this(true);
      }
  
      public FlowControlHandler(boolean releaseMessages) {
          this.releaseMessages = releaseMessages;
      }
  
      /**
       * Determine if the underlying {@link Queue} is empty. This method exists for
       * testing, debugging and inspection purposes and it is not Thread safe!
       */
      boolean isQueueEmpty() {
          return queue == null || queue.isEmpty();
      }
  
      /**
       * Releases all messages and destroys the {@link Queue}.
       */
      private void destroy() {
          if (queue != null) {
 
             if (!queue.isEmpty()) {
                 logger.trace("Non-empty queue: {}", queue);
 
                 if (releaseMessages) {
                     Object msg;
                     while ((msg = queue.poll()) != null) {
                         ReferenceCountUtil.safeRelease(msg);
                     }
                 }
             }
 
             queue.recycle();
             queue = null;
         }
     }
 
     @Override
     public void handlerAdded(ChannelHandlerContext ctx) throws Exception {
         config = ctx.channel().config();
     }
 
     @Override
     public void handlerRemoved(ChannelHandlerContext ctx) throws Exception {
         super.handlerRemoved(ctx);
         if (!isQueueEmpty()) {
             dequeue(ctx, queue.size());
         }
         destroy();
     }
 
     @Override
     public void channelInactive(ChannelHandlerContext ctx) throws Exception {
         destroy();
         ctx.fireChannelInactive();
     }
 
     @Override
     public void read(ChannelHandlerContext ctx) throws Exception {
         if (dequeue(ctx, 1) == 0) {
             // It seems no messages were consumed. We need to read() some
             // messages from upstream and once one arrives it need to be
             // relayed to downstream to keep the flow going.
             shouldConsume = true;
             ctx.read();
         } else if (config.isAutoRead()) {
             ctx.read();
         }
     }
 
     @Override
     public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
         if (queue == null) {
             queue = RecyclableArrayDeque.newInstance();
         }
 
         queue.offer(msg);
 
         // We just received one message. Do we need to relay it regardless
         // of the auto reading configuration? The answer is yes if this
         // method was called as a result of a prior read() call.
         int minConsume = shouldConsume ? 1 : 0;
         shouldConsume = false;
 
         dequeue(ctx, minConsume);
     }
 
     @Override
     public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
         if (isQueueEmpty()) {
             ctx.fireChannelReadComplete();
         } else {
             // Don't relay completion events from upstream as they
             // make no sense in this context. See dequeue() where
             // a new set of completion events is being produced.
         }
     }
 
     /**
      * Dequeues one or many (or none) messages depending on the channel's auto
      * reading state and returns the number of messages that were consumed from
      * the internal queue.
      *
      * The {@code minConsume} argument is used to force {@code dequeue()} into
      * consuming that number of messages regardless of the channel's auto
      * reading configuration.
      *
      * @see #read(ChannelHandlerContext)
      * @see #channelRead(ChannelHandlerContext, Object)
      */
     private int dequeue(ChannelHandlerContext ctx, int minConsume) {
         int consumed = 0;
 
         // fireChannelRead(...) may call ctx.read() and so this method may reentrance. Because of this we need to
         // check if queue was set to null in the meantime and if so break the loop.
         while (queue != null && (consumed < minConsume || config.isAutoRead())) {
             Object msg = queue.poll();
             if (msg == null) {
                 break;
             }
 
             ++consumed;
             ctx.fireChannelRead(msg);
         }
 
         // We're firing a completion event every time one (or more)
         // messages were consumed and the queue ended up being drained
         // to an empty state.
         if (queue != null && queue.isEmpty()) {
             queue.recycle();
             queue = null;
 
             if (consumed > 0) {
                 ctx.fireChannelReadComplete();
             }
         }
 
         return consumed;
     }
 
     /**
      * A recyclable {@link ArrayDeque}.
      */
     private static final class RecyclableArrayDeque extends ArrayDeque<Object> {
 
         private static final long serialVersionUID = 0L;
 
         /**
          * A value of {@code 2} should be a good choice for most scenarios.
          */
         private static final int DEFAULT_NUM_ELEMENTS = 2;
 
         private static final ObjectPool<RecyclableArrayDeque> RECYCLER = ObjectPool.newPool(
                 new ObjectCreator<RecyclableArrayDeque>() {
             @Override
             public RecyclableArrayDeque newObject(Handle<RecyclableArrayDeque> handle) {
                 return new RecyclableArrayDeque(DEFAULT_NUM_ELEMENTS, handle);
             }
         });
 
         public static RecyclableArrayDeque newInstance() {
             return RECYCLER.get();
         }
 
         private final Handle<RecyclableArrayDeque> handle;
 
         private RecyclableArrayDeque(int numElements, Handle<RecyclableArrayDeque> handle) {
             super(numElements);
             this.handle = handle;
         }
 
         public void recycle() {
             clear();
             handle.recycle(this);
         }
     }
 }