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 if (isFull) {
159 out.add(ReferenceCountUtil.retain(res));
160 } else {
161 out.add(ReferenceCountUtil.retain(res));
162
163 state = State.PASS_THROUGH;
164 }
165 break;
166 }
167
168 if (isFull) {
169
170 if (!((ByteBufHolder) res).content().isReadable()) {
171 out.add(ReferenceCountUtil.retain(res));
172 break;
173 }
174 }
175
176
177 final Result result = beginEncode(res, acceptEncoding.toString());
178
179
180 if (result == null) {
181 if (isFull) {
182 out.add(ReferenceCountUtil.retain(res));
183 } else {
184 out.add(ReferenceCountUtil.retain(res));
185
186 state = State.PASS_THROUGH;
187 }
188 break;
189 }
190
191 encoder = result.contentEncoder();
192
193
194
195 res.headers().set(HttpHeaderNames.CONTENT_ENCODING, result.targetContentEncoding());
196
197
198 if (isFull) {
199
200 HttpResponse newRes = new DefaultHttpResponse(res.protocolVersion(), res.status());
201 newRes.headers().set(res.headers());
202 out.add(newRes);
203
204 ensureContent(res);
205 encodeFullResponse(newRes, (HttpContent) res, out);
206 break;
207 } else {
208
209 res.headers().remove(HttpHeaderNames.CONTENT_LENGTH);
210 res.headers().set(HttpHeaderNames.TRANSFER_ENCODING, HttpHeaderValues.CHUNKED);
211
212 out.add(ReferenceCountUtil.retain(res));
213 state = State.AWAIT_CONTENT;
214 if (!(msg instanceof HttpContent)) {
215
216
217 break;
218 }
219
220 }
221 }
222 case AWAIT_CONTENT: {
223 ensureContent(msg);
224 if (encodeContent((HttpContent) msg, out)) {
225 state = State.AWAIT_HEADERS;
226 } else if (out.isEmpty()) {
227
228 out.add(new DefaultHttpContent(Unpooled.EMPTY_BUFFER));
229 }
230 break;
231 }
232 case PASS_THROUGH: {
233 ensureContent(msg);
234 out.add(ReferenceCountUtil.retain(msg));
235
236 if (msg instanceof LastHttpContent) {
237 state = State.AWAIT_HEADERS;
238 }
239 break;
240 }
241 }
242 }
243
244 private void encodeFullResponse(HttpResponse newRes, HttpContent content, List<Object> out) {
245 int existingMessages = out.size();
246 encodeContent(content, out);
247
248 if (HttpUtil.isContentLengthSet(newRes)) {
249
250 int messageSize = 0;
251 for (int i = existingMessages; i < out.size(); i++) {
252 Object item = out.get(i);
253 if (item instanceof HttpContent) {
254 messageSize += ((HttpContent) item).content().readableBytes();
255 }
256 }
257 HttpUtil.setContentLength(newRes, messageSize);
258 } else {
259 newRes.headers().set(HttpHeaderNames.TRANSFER_ENCODING, HttpHeaderValues.CHUNKED);
260 }
261 }
262
263 private static boolean isPassthru(HttpVersion version, int code, CharSequence httpMethod) {
264 return code < 200 || code == 204 || code == 304 ||
265 (httpMethod == ZERO_LENGTH_HEAD || (httpMethod == ZERO_LENGTH_CONNECT && code == 200)) ||
266 version == HttpVersion.HTTP_1_0;
267 }
268
269 private static void ensureHeaders(HttpObject msg) {
270 if (!(msg instanceof HttpResponse)) {
271 throw new IllegalStateException(
272 "unexpected message type: " +
273 msg.getClass().getName() + " (expected: " + HttpResponse.class.getSimpleName() + ')');
274 }
275 }
276
277 private static void ensureContent(HttpObject msg) {
278 if (!(msg instanceof HttpContent)) {
279 throw new IllegalStateException(
280 "unexpected message type: " +
281 msg.getClass().getName() + " (expected: " + HttpContent.class.getSimpleName() + ')');
282 }
283 }
284
285 private boolean encodeContent(HttpContent c, List<Object> out) {
286 ByteBuf content = c.content();
287
288 encode(content, out);
289
290 if (c instanceof LastHttpContent) {
291 finishEncode(out);
292 LastHttpContent last = (LastHttpContent) c;
293
294
295
296 HttpHeaders headers = last.trailingHeaders();
297 if (headers.isEmpty()) {
298 out.add(LastHttpContent.EMPTY_LAST_CONTENT);
299 } else {
300 out.add(new ComposedLastHttpContent(headers, DecoderResult.SUCCESS));
301 }
302 return true;
303 }
304 return false;
305 }
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321 protected abstract Result beginEncode(HttpResponse httpResponse, String acceptEncoding) throws Exception;
322
323 @Override
324 public void handlerRemoved(ChannelHandlerContext ctx) throws Exception {
325 cleanupSafely(ctx);
326 super.handlerRemoved(ctx);
327 }
328
329 @Override
330 public void channelInactive(ChannelHandlerContext ctx) throws Exception {
331 cleanupSafely(ctx);
332 super.channelInactive(ctx);
333 }
334
335 private void cleanup() {
336 if (encoder != null) {
337
338 encoder.finishAndReleaseAll();
339 encoder = null;
340 }
341 }
342
343 private void cleanupSafely(ChannelHandlerContext ctx) {
344 try {
345 cleanup();
346 } catch (Throwable cause) {
347
348
349 ctx.fireExceptionCaught(cause);
350 }
351 }
352
353 private void encode(ByteBuf in, List<Object> out) {
354
355 encoder.writeOutbound(in.retain());
356 fetchEncoderOutput(out);
357 }
358
359 private void finishEncode(List<Object> out) {
360 if (encoder.finish()) {
361 fetchEncoderOutput(out);
362 }
363 encoder = null;
364 }
365
366 private void fetchEncoderOutput(List<Object> out) {
367 for (;;) {
368 ByteBuf buf = encoder.readOutbound();
369 if (buf == null) {
370 break;
371 }
372 if (!buf.isReadable()) {
373 buf.release();
374 continue;
375 }
376 out.add(new DefaultHttpContent(buf));
377 }
378 }
379
380 public static final class Result {
381 private final String targetContentEncoding;
382 private final EmbeddedChannel contentEncoder;
383
384 public Result(String targetContentEncoding, EmbeddedChannel contentEncoder) {
385 this.targetContentEncoding = ObjectUtil.checkNotNull(targetContentEncoding, "targetContentEncoding");
386 this.contentEncoder = ObjectUtil.checkNotNull(contentEncoder, "contentEncoder");
387 }
388
389 public String targetContentEncoding() {
390 return targetContentEncoding;
391 }
392
393 public EmbeddedChannel contentEncoder() {
394 return contentEncoder;
395 }
396 }
397 }