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