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