1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16 package io.netty.handler.codec.http;
17
18 import io.netty.buffer.ByteBuf;
19 import io.netty.buffer.ByteBufHolder;
20 import io.netty.buffer.Unpooled;
21 import io.netty.channel.ChannelHandlerContext;
22 import io.netty.channel.embedded.EmbeddedChannel;
23 import io.netty.handler.codec.DecoderResult;
24 import io.netty.handler.codec.MessageToMessageCodec;
25 import io.netty.util.ReferenceCountUtil;
26 import io.netty.util.internal.ObjectUtil;
27 import io.netty.util.internal.StringUtil;
28
29 import java.util.ArrayDeque;
30 import java.util.List;
31 import java.util.Queue;
32
33 import static io.netty.handler.codec.http.HttpHeaderNames.*;
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57 public abstract class HttpContentEncoder extends MessageToMessageCodec<HttpRequest, HttpObject> {
58 public static final int DEFAULT_MAX_PIPELINE_DEPTH = 128;
59
60 private enum State {
61 PASS_THROUGH,
62 AWAIT_HEADERS,
63 AWAIT_CONTENT
64 }
65
66 private static final CharSequence ZERO_LENGTH_HEAD = "HEAD";
67 private static final CharSequence ZERO_LENGTH_CONNECT = "CONNECT";
68
69 private final int maxPipelineDepth;
70 private final Queue<CharSequence> acceptEncodingQueue = new ArrayDeque<CharSequence>();
71 private EmbeddedChannel encoder;
72 private State state = State.AWAIT_HEADERS;
73
74 public HttpContentEncoder() {
75 this(DEFAULT_MAX_PIPELINE_DEPTH);
76 }
77
78 public HttpContentEncoder(int maxPipelineDepth) {
79 super(HttpRequest.class, HttpObject.class);
80 this.maxPipelineDepth = ObjectUtil.checkPositive(maxPipelineDepth, "maxPipelineDepth");
81 }
82
83 @Override
84 public boolean acceptOutboundMessage(Object msg) throws Exception {
85 return msg instanceof HttpContent || msg instanceof HttpResponse;
86 }
87
88 @Override
89 protected void decode(ChannelHandlerContext ctx, HttpRequest msg, List<Object> out) throws Exception {
90 if (maxPipelineDepth <= acceptEncodingQueue.size()) {
91 throw new IllegalStateException("maxPipelineDepth exceeded: " + maxPipelineDepth);
92 }
93 CharSequence acceptEncoding;
94 List<String> acceptEncodingHeaders = msg.headers().getAll(ACCEPT_ENCODING);
95 switch (acceptEncodingHeaders.size()) {
96 case 0:
97 acceptEncoding = HttpContentDecoder.IDENTITY;
98 break;
99 case 1:
100 acceptEncoding = acceptEncodingHeaders.get(0);
101 break;
102 default:
103
104 acceptEncoding = StringUtil.join(",", acceptEncodingHeaders);
105 break;
106 }
107
108 HttpMethod method = msg.method();
109 if (HttpMethod.HEAD.equals(method)) {
110 acceptEncoding = ZERO_LENGTH_HEAD;
111 } else if (HttpMethod.CONNECT.equals(method)) {
112 acceptEncoding = ZERO_LENGTH_CONNECT;
113 }
114
115 acceptEncodingQueue.add(acceptEncoding);
116 out.add(ReferenceCountUtil.retain(msg));
117 }
118
119 @Override
120 protected void encode(ChannelHandlerContext ctx, HttpObject msg, List<Object> out) throws Exception {
121 final boolean isFull = msg instanceof HttpResponse && msg instanceof LastHttpContent;
122 switch (state) {
123 case AWAIT_HEADERS: {
124 ensureHeaders(msg);
125 assert encoder == null;
126
127 final HttpResponse res = (HttpResponse) msg;
128 final int code = res.status().code();
129 final HttpStatusClass codeClass = res.status().codeClass();
130 final CharSequence acceptEncoding;
131 if (codeClass == HttpStatusClass.INFORMATIONAL) {
132
133
134
135 acceptEncoding = null;
136 } else {
137
138 acceptEncoding = acceptEncodingQueue.poll();
139 if (acceptEncoding == null) {
140 throw new IllegalStateException("cannot send more responses than requests");
141 }
142 }
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157 if (isPassthru(res.protocolVersion(), code, acceptEncoding)) {
158 out.add(ReferenceCountUtil.retain(res));
159 if (!isFull) {
160
161 state = State.PASS_THROUGH;
162 }
163 break;
164 }
165
166 if (isFull) {
167
168 if (!((ByteBufHolder) res).content().isReadable()) {
169 out.add(ReferenceCountUtil.retain(res));
170 break;
171 }
172 }
173
174
175 final Result result = beginEncode(res, acceptEncoding.toString());
176
177
178 if (result == null) {
179 out.add(ReferenceCountUtil.retain(res));
180 if (!isFull) {
181
182 state = State.PASS_THROUGH;
183 }
184 break;
185 }
186
187 EmbeddedChannel contentEncoder = result.contentEncoder();
188 try {
189
190
191 res.headers().set(HttpHeaderNames.CONTENT_ENCODING, result.targetContentEncoding());
192
193
194 if (isFull) {
195
196 HttpResponse newRes = new DefaultHttpResponse(res.protocolVersion(), res.status());
197 newRes.headers().set(res.headers());
198 out.add(newRes);
199
200 ensureContent(res);
201 encoder = contentEncoder;
202 encodeFullResponse(newRes, (HttpContent) res, out);
203 contentEncoder = null;
204 break;
205 } else {
206
207 res.headers().remove(HttpHeaderNames.CONTENT_LENGTH);
208 res.headers().set(HttpHeaderNames.TRANSFER_ENCODING, HttpHeaderValues.CHUNKED);
209
210 out.add(ReferenceCountUtil.retain(res));
211 state = State.AWAIT_CONTENT;
212 encoder = contentEncoder;
213 contentEncoder = null;
214 if (!(msg instanceof HttpContent)) {
215
216
217 break;
218 }
219
220 }
221 } finally {
222 if (contentEncoder != null) {
223 if (encoder == contentEncoder) {
224 encoder = null;
225 }
226 contentEncoder.finishAndReleaseAll();
227 }
228 }
229 }
230 case AWAIT_CONTENT: {
231 ensureContent(msg);
232 if (encodeContent((HttpContent) msg, out)) {
233 state = State.AWAIT_HEADERS;
234 } else if (out.isEmpty()) {
235
236 out.add(new DefaultHttpContent(Unpooled.EMPTY_BUFFER));
237 }
238 break;
239 }
240 case PASS_THROUGH: {
241 ensureContent(msg);
242 out.add(ReferenceCountUtil.retain(msg));
243
244 if (msg instanceof LastHttpContent) {
245 state = State.AWAIT_HEADERS;
246 }
247 break;
248 }
249 }
250 }
251
252 private void encodeFullResponse(HttpResponse newRes, HttpContent content, List<Object> out) {
253 int existingMessages = out.size();
254 encodeContent(content, out);
255
256 if (HttpUtil.isContentLengthSet(newRes)) {
257
258 int messageSize = 0;
259 for (int i = existingMessages; i < out.size(); i++) {
260 Object item = out.get(i);
261 if (item instanceof HttpContent) {
262 messageSize += ((HttpContent) item).content().readableBytes();
263 }
264 }
265 HttpUtil.setContentLength(newRes, messageSize);
266 } else {
267 newRes.headers().set(HttpHeaderNames.TRANSFER_ENCODING, HttpHeaderValues.CHUNKED);
268 }
269 }
270
271 private static boolean isPassthru(HttpVersion version, int code, CharSequence httpMethod) {
272 return code < 200 || code == 204 || code == 304 ||
273 (httpMethod == ZERO_LENGTH_HEAD || (httpMethod == ZERO_LENGTH_CONNECT && code == 200)) ||
274 version == HttpVersion.HTTP_1_0;
275 }
276
277 private static void ensureHeaders(HttpObject msg) {
278 if (!(msg instanceof HttpResponse)) {
279 throw new IllegalStateException(
280 "unexpected message type: " +
281 msg.getClass().getName() + " (expected: " + HttpResponse.class.getSimpleName() + ')');
282 }
283 }
284
285 private static void ensureContent(HttpObject msg) {
286 if (!(msg instanceof HttpContent)) {
287 throw new IllegalStateException(
288 "unexpected message type: " +
289 msg.getClass().getName() + " (expected: " + HttpContent.class.getSimpleName() + ')');
290 }
291 }
292
293 private boolean encodeContent(HttpContent c, List<Object> out) {
294 ByteBuf content = c.content();
295
296 encode(content, out);
297
298 if (c instanceof LastHttpContent) {
299 finishEncode(out);
300 LastHttpContent last = (LastHttpContent) c;
301
302
303
304 HttpHeaders headers = last.trailingHeaders();
305 if (headers.isEmpty()) {
306 out.add(LastHttpContent.EMPTY_LAST_CONTENT);
307 } else {
308 out.add(new ComposedLastHttpContent(headers, DecoderResult.SUCCESS));
309 }
310 return true;
311 }
312 return false;
313 }
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329 protected abstract Result beginEncode(HttpResponse httpResponse, String acceptEncoding) throws Exception;
330
331 @Override
332 public void handlerRemoved(ChannelHandlerContext ctx) throws Exception {
333 cleanupSafely(ctx);
334 super.handlerRemoved(ctx);
335 }
336
337 @Override
338 public void channelInactive(ChannelHandlerContext ctx) throws Exception {
339 cleanupSafely(ctx);
340 super.channelInactive(ctx);
341 }
342
343 private void cleanup() {
344 if (encoder != null) {
345
346 encoder.finishAndReleaseAll();
347 encoder = null;
348 }
349 }
350
351 private void cleanupSafely(ChannelHandlerContext ctx) {
352 try {
353 cleanup();
354 } catch (Throwable cause) {
355
356
357 ctx.fireExceptionCaught(cause);
358 }
359 }
360
361 private void encode(ByteBuf in, List<Object> out) {
362
363 encoder.writeOutbound(in.retain());
364 fetchEncoderOutput(out);
365 }
366
367 private void finishEncode(List<Object> out) {
368 if (encoder.finish()) {
369 fetchEncoderOutput(out);
370 }
371 encoder = null;
372 }
373
374 private void fetchEncoderOutput(List<Object> out) {
375 for (;;) {
376 ByteBuf buf = encoder.readOutbound();
377 if (buf == null) {
378 break;
379 }
380 if (!buf.isReadable()) {
381 buf.release();
382 continue;
383 }
384 out.add(new DefaultHttpContent(buf));
385 }
386 }
387
388 public static final class Result {
389 private final String targetContentEncoding;
390 private final EmbeddedChannel contentEncoder;
391
392 public Result(String targetContentEncoding, EmbeddedChannel contentEncoder) {
393 this.targetContentEncoding = ObjectUtil.checkNotNull(targetContentEncoding, "targetContentEncoding");
394 this.contentEncoder = ObjectUtil.checkNotNull(contentEncoder, "contentEncoder");
395 }
396
397 public String targetContentEncoding() {
398 return targetContentEncoding;
399 }
400
401 public EmbeddedChannel contentEncoder() {
402 return contentEncoder;
403 }
404 }
405 }