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