View Javadoc
1   /*
2    * Copyright 2016 The Netty Project
3    *
4    * The Netty Project licenses this file to you under the Apache License, version
5    * 2.0 (the "License"); you may not use this file except in compliance with the
6    * License. You may obtain a copy of the License at:
7    *
8    * http://www.apache.org/licenses/LICENSE-2.0
9    *
10   * Unless required by applicable law or agreed to in writing, software
11   * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
12   * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
13   * License for the specific language governing permissions and limitations under
14   * the License.
15   */
16  package io.netty.handler.flow;
17  
18  import java.util.ArrayDeque;
19  import java.util.Queue;
20  
21  import io.netty.channel.ChannelConfig;
22  import io.netty.channel.ChannelDuplexHandler;
23  import io.netty.channel.ChannelHandler;
24  import io.netty.channel.ChannelHandlerContext;
25  import io.netty.handler.codec.ByteToMessageDecoder;
26  import io.netty.handler.codec.MessageToByteEncoder;
27  import io.netty.util.Recycler;
28  import io.netty.util.Recycler.Handle;
29  import io.netty.util.ReferenceCountUtil;
30  import io.netty.util.internal.logging.InternalLogger;
31  import io.netty.util.internal.logging.InternalLoggerFactory;
32  
33  /**
34   * The {@link FlowControlHandler} ensures that only one message per {@code read()} is sent downstream.
35   *
36   * Classes such as {@link ByteToMessageDecoder} or {@link MessageToByteEncoder} are free to emit as
37   * many events as they like for any given input. A channel's auto reading configuration doesn't usually
38   * apply in these scenarios. This is causing problems in downstream {@link ChannelHandler}s that would
39   * like to hold subsequent events while they're processing one event. It's a common problem with the
40   * {@code HttpObjectDecoder} that will very often fire a {@code HttpRequest} that is immediately followed
41   * by a {@code LastHttpContent} event.
42   *
43   * <pre>{@code
44   * ChannelPipeline pipeline = ...;
45   *
46   * pipeline.addLast(new HttpServerCodec());
47   * pipeline.addLast(new FlowControlHandler());
48   *
49   * pipeline.addLast(new MyExampleHandler());
50   *
51   * class MyExampleHandler extends ChannelInboundHandlerAdapter {
52   *   @Override
53   *   public void channelRead(ChannelHandlerContext ctx, Object msg) {
54   *     if (msg instanceof HttpRequest) {
55   *       ctx.channel().config().setAutoRead(false);
56   *
57   *       // The FlowControlHandler will hold any subsequent events that
58   *       // were emitted by HttpObjectDecoder until auto reading is turned
59   *       // back on or Channel#read() is being called.
60   *     }
61   *   }
62   * }
63   * }</pre>
64   *
65   * @see ChannelConfig#setAutoRead(boolean)
66   */
67  public class FlowControlHandler extends ChannelDuplexHandler {
68      private static final InternalLogger logger = InternalLoggerFactory.getInstance(FlowControlHandler.class);
69  
70      private final boolean releaseMessages;
71  
72      private RecyclableArrayDeque queue;
73  
74      private ChannelConfig config;
75  
76      private boolean shouldConsume;
77  
78      public FlowControlHandler() {
79          this(true);
80      }
81  
82      public FlowControlHandler(boolean releaseMessages) {
83          this.releaseMessages = releaseMessages;
84      }
85  
86      /**
87       * Determine if the underlying {@link Queue} is empty. This method exists for
88       * testing, debugging and inspection purposes and it is not Thread safe!
89       */
90      boolean isQueueEmpty() {
91          return queue.isEmpty();
92      }
93  
94      /**
95       * Releases all messages and destroys the {@link Queue}.
96       */
97      private void destroy() {
98          if (queue != null) {
99  
100             if (!queue.isEmpty()) {
101                 logger.trace("Non-empty queue: {}", queue);
102 
103                 if (releaseMessages) {
104                     Object msg;
105                     while ((msg = queue.poll()) != null) {
106                         ReferenceCountUtil.safeRelease(msg);
107                     }
108                 }
109             }
110 
111             queue.recycle();
112             queue = null;
113         }
114     }
115 
116     @Override
117     public void handlerAdded(ChannelHandlerContext ctx) throws Exception {
118         config = ctx.channel().config();
119     }
120 
121     @Override
122     public void channelInactive(ChannelHandlerContext ctx) throws Exception {
123         destroy();
124         ctx.fireChannelInactive();
125     }
126 
127     @Override
128     public void read(ChannelHandlerContext ctx) throws Exception {
129         if (dequeue(ctx, 1) == 0) {
130             // It seems no messages were consumed. We need to read() some
131             // messages from upstream and once one arrives it need to be
132             // relayed to downstream to keep the flow going.
133             shouldConsume = true;
134             ctx.read();
135         }
136     }
137 
138     @Override
139     public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
140         if (queue == null) {
141             queue = RecyclableArrayDeque.newInstance();
142         }
143 
144         queue.offer(msg);
145 
146         // We just received one message. Do we need to relay it regardless
147         // of the auto reading configuration? The answer is yes if this
148         // method was called as a result of a prior read() call.
149         int minConsume = shouldConsume ? 1 : 0;
150         shouldConsume = false;
151 
152         dequeue(ctx, minConsume);
153     }
154 
155     @Override
156     public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
157         // Don't relay completion events from upstream as they
158         // make no sense in this context. See dequeue() where
159         // a new set of completion events is being produced.
160     }
161 
162     /**
163      * Dequeues one or many (or none) messages depending on the channel's auto
164      * reading state and returns the number of messages that were consumed from
165      * the internal queue.
166      *
167      * The {@code minConsume} argument is used to force {@code dequeue()} into
168      * consuming that number of messages regardless of the channel's auto
169      * reading configuration.
170      *
171      * @see #read(ChannelHandlerContext)
172      * @see #channelRead(ChannelHandlerContext, Object)
173      */
174     private int dequeue(ChannelHandlerContext ctx, int minConsume) {
175         if (queue != null) {
176 
177             int consumed = 0;
178 
179             Object msg;
180             while ((consumed < minConsume) || config.isAutoRead()) {
181                 msg = queue.poll();
182                 if (msg == null) {
183                     break;
184                 }
185 
186                 ++consumed;
187                 ctx.fireChannelRead(msg);
188             }
189 
190             // We're firing a completion event every time one (or more)
191             // messages were consumed and the queue ended up being drained
192             // to an empty state.
193             if (queue.isEmpty() && consumed > 0) {
194                 ctx.fireChannelReadComplete();
195             }
196 
197             return consumed;
198         }
199 
200         return 0;
201     }
202 
203     /**
204      * A recyclable {@link ArrayDeque}.
205      */
206     private static final class RecyclableArrayDeque extends ArrayDeque<Object> {
207 
208         private static final long serialVersionUID = 0L;
209 
210         /**
211          * A value of {@code 2} should be a good choice for most scenarios.
212          */
213         private static final int DEFAULT_NUM_ELEMENTS = 2;
214 
215         private static final Recycler<RecyclableArrayDeque> RECYCLER = new Recycler<RecyclableArrayDeque>() {
216             @Override
217             protected RecyclableArrayDeque newObject(Handle<RecyclableArrayDeque> handle) {
218                 return new RecyclableArrayDeque(DEFAULT_NUM_ELEMENTS, handle);
219             }
220         };
221 
222         public static RecyclableArrayDeque newInstance() {
223             return RECYCLER.get();
224         }
225 
226         private final Handle<RecyclableArrayDeque> handle;
227 
228         private RecyclableArrayDeque(int numElements, Handle<RecyclableArrayDeque> handle) {
229             super(numElements);
230             this.handle = handle;
231         }
232 
233         public void recycle() {
234             clear();
235             handle.recycle(this);
236         }
237     }
238 }