1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19 package org.eclipse.jetty.client;
20
21 import java.io.EOFException;
22 import java.io.IOException;
23 import java.net.URI;
24 import java.nio.ByteBuffer;
25 import java.util.Collections;
26 import java.util.Enumeration;
27 import java.util.HashMap;
28 import java.util.List;
29 import java.util.Map;
30 import java.util.concurrent.TimeoutException;
31 import java.util.concurrent.atomic.AtomicMarkableReference;
32 import java.util.concurrent.atomic.AtomicReference;
33
34 import org.eclipse.jetty.client.api.Response;
35 import org.eclipse.jetty.client.api.Result;
36 import org.eclipse.jetty.http.HttpField;
37 import org.eclipse.jetty.http.HttpHeader;
38 import org.eclipse.jetty.http.HttpMethod;
39 import org.eclipse.jetty.http.HttpParser;
40 import org.eclipse.jetty.http.HttpVersion;
41 import org.eclipse.jetty.io.ByteBufferPool;
42 import org.eclipse.jetty.io.EndPoint;
43 import org.eclipse.jetty.io.EofException;
44 import org.eclipse.jetty.util.log.Log;
45 import org.eclipse.jetty.util.log.Logger;
46
47 public class HttpReceiver implements HttpParser.ResponseHandler<ByteBuffer>
48 {
49 private static final Logger LOG = Log.getLogger(HttpReceiver.class);
50
51 private final AtomicReference<State> state = new AtomicReference<>(State.IDLE);
52 private final HttpParser parser = new HttpParser(this);
53 private final HttpConnection connection;
54 private ContentDecoder decoder;
55
56 public HttpReceiver(HttpConnection connection)
57 {
58 this.connection = connection;
59 }
60
61 public void receive()
62 {
63 EndPoint endPoint = connection.getEndPoint();
64 HttpClient client = connection.getHttpClient();
65 ByteBufferPool bufferPool = client.getByteBufferPool();
66 ByteBuffer buffer = bufferPool.acquire(client.getResponseBufferSize(), true);
67 try
68 {
69 while (true)
70 {
71 int read = endPoint.fill(buffer);
72 LOG.debug("Read {} bytes from {}", read, connection);
73 if (read > 0)
74 {
75 parse(buffer);
76 }
77 else if (read == 0)
78 {
79 fillInterested();
80 break;
81 }
82 else
83 {
84 shutdown();
85 break;
86 }
87 }
88 }
89 catch (EofException x)
90 {
91 LOG.ignore(x);
92 failAndClose(x);
93 }
94 catch (Exception x)
95 {
96 LOG.debug(x);
97 failAndClose(x);
98 }
99 finally
100 {
101 bufferPool.release(buffer);
102 }
103 }
104
105 private void parse(ByteBuffer buffer)
106 {
107 while (buffer.hasRemaining())
108 parser.parseNext(buffer);
109 }
110
111 private void fillInterested()
112 {
113 State state = this.state.get();
114 if (state == State.IDLE || state == State.RECEIVE)
115 connection.fillInterested();
116 }
117
118 private void shutdown()
119 {
120
121 parser.shutdownInput();
122 State state = this.state.get();
123 if (state == State.IDLE || state == State.RECEIVE)
124 {
125 if (!fail(new EOFException()))
126 connection.close();
127 }
128 }
129
130
131 @Override
132 public int getHeaderCacheSize()
133 {
134
135 return 256;
136 }
137
138 @Override
139 public boolean startResponse(HttpVersion version, int status, String reason)
140 {
141 if (updateState(State.IDLE, State.RECEIVE))
142 {
143 HttpExchange exchange = connection.getExchange();
144
145 if (exchange != null)
146 {
147 HttpConversation conversation = exchange.getConversation();
148 HttpResponse response = exchange.getResponse();
149
150 parser.setHeadResponse(exchange.getRequest().getMethod() == HttpMethod.HEAD);
151 response.version(version).status(status).reason(reason);
152
153
154 HttpClient client = connection.getHttpClient();
155 ProtocolHandler protocolHandler = client.findProtocolHandler(exchange.getRequest(), response);
156 Response.Listener handlerListener = null;
157 if (protocolHandler != null)
158 {
159 handlerListener = protocolHandler.getResponseListener();
160 LOG.debug("Found protocol handler {}", protocolHandler);
161 }
162 exchange.getConversation().updateResponseListeners(handlerListener);
163
164 LOG.debug("Receiving {}", response);
165 ResponseNotifier notifier = connection.getDestination().getResponseNotifier();
166 notifier.notifyBegin(conversation.getResponseListeners(), response);
167 }
168 }
169 return false;
170 }
171
172 @Override
173 public boolean parsedHeader(HttpField field)
174 {
175 if (updateState(State.RECEIVE, State.RECEIVE))
176 {
177 HttpExchange exchange = connection.getExchange();
178
179 if (exchange != null)
180 {
181 HttpConversation conversation = exchange.getConversation();
182 HttpResponse response = exchange.getResponse();
183 ResponseNotifier notifier = connection.getDestination().getResponseNotifier();
184 boolean process = notifier.notifyHeader(conversation.getResponseListeners(), response, field);
185 if (process)
186 {
187 response.getHeaders().add(field);
188 HttpHeader fieldHeader = field.getHeader();
189 if (fieldHeader != null)
190 {
191 switch (fieldHeader)
192 {
193 case SET_COOKIE:
194 case SET_COOKIE2:
195 {
196 storeCookie(exchange.getRequest().getURI(), field);
197 break;
198 }
199 default:
200 {
201 break;
202 }
203 }
204 }
205 }
206 }
207 }
208 return false;
209 }
210
211 private void storeCookie(URI uri, HttpField field)
212 {
213 try
214 {
215 String value = field.getValue();
216 if (value != null)
217 {
218 Map<String, List<String>> header = new HashMap<>(1);
219 header.put(field.getHeader().asString(), Collections.singletonList(value));
220 connection.getHttpClient().getCookieManager().put(uri, header);
221 }
222 }
223 catch (IOException x)
224 {
225 LOG.debug(x);
226 }
227 }
228
229 @Override
230 public boolean headerComplete()
231 {
232 if (updateState(State.RECEIVE, State.RECEIVE))
233 {
234 HttpExchange exchange = connection.getExchange();
235
236 if (exchange != null)
237 {
238 HttpConversation conversation = exchange.getConversation();
239 HttpResponse response = exchange.getResponse();
240 LOG.debug("Headers {}", response);
241 ResponseNotifier notifier = connection.getDestination().getResponseNotifier();
242 notifier.notifyHeaders(conversation.getResponseListeners(), response);
243
244 Enumeration<String> contentEncodings = response.getHeaders().getValues(HttpHeader.CONTENT_ENCODING.asString(), ",");
245 if (contentEncodings != null)
246 {
247 for (ContentDecoder.Factory factory : connection.getHttpClient().getContentDecoderFactories())
248 {
249 while (contentEncodings.hasMoreElements())
250 {
251 if (factory.getEncoding().equalsIgnoreCase(contentEncodings.nextElement()))
252 {
253 this.decoder = factory.newContentDecoder();
254 break;
255 }
256 }
257 }
258 }
259 }
260 }
261 return false;
262 }
263
264 @Override
265 public boolean content(ByteBuffer buffer)
266 {
267 if (updateState(State.RECEIVE, State.RECEIVE))
268 {
269 HttpExchange exchange = connection.getExchange();
270
271 if (exchange != null)
272 {
273 HttpConversation conversation = exchange.getConversation();
274 HttpResponse response = exchange.getResponse();
275 LOG.debug("Content {}: {} bytes", response, buffer.remaining());
276
277 ContentDecoder decoder = this.decoder;
278 if (decoder != null)
279 {
280 buffer = decoder.decode(buffer);
281 LOG.debug("{} {}: {} bytes", decoder, response, buffer.remaining());
282 }
283
284 ResponseNotifier notifier = connection.getDestination().getResponseNotifier();
285 notifier.notifyContent(conversation.getResponseListeners(), response, buffer);
286 }
287 }
288 return false;
289 }
290
291 @Override
292 public boolean messageComplete()
293 {
294 if (updateState(State.RECEIVE, State.RECEIVE))
295 success();
296 return true;
297 }
298
299 protected boolean success()
300 {
301 HttpExchange exchange = connection.getExchange();
302 if (exchange == null)
303 return false;
304
305 AtomicMarkableReference<Result> completion = exchange.responseComplete(null);
306 if (!completion.isMarked())
307 return false;
308
309 parser.reset();
310 decoder = null;
311
312 if (!updateState(State.RECEIVE, State.IDLE))
313 throw new IllegalStateException();
314
315 exchange.terminateResponse();
316
317 HttpResponse response = exchange.getResponse();
318 List<Response.ResponseListener> listeners = exchange.getConversation().getResponseListeners();
319 ResponseNotifier notifier = connection.getDestination().getResponseNotifier();
320 notifier.notifySuccess(listeners, response);
321 LOG.debug("Received {}", response);
322
323 Result result = completion.getReference();
324 if (result != null)
325 {
326 connection.complete(exchange, !result.isFailed());
327 notifier.notifyComplete(listeners, result);
328 }
329
330 return true;
331 }
332
333 protected boolean fail(Throwable failure)
334 {
335 HttpExchange exchange = connection.getExchange();
336
337
338
339
340 if (exchange == null)
341 return false;
342
343 AtomicMarkableReference<Result> completion = exchange.responseComplete(failure);
344 if (!completion.isMarked())
345 return false;
346
347 parser.close();
348 decoder = null;
349
350 while (true)
351 {
352 State current = state.get();
353 if (updateState(current, State.FAILURE))
354 break;
355 }
356
357 exchange.terminateResponse();
358
359 HttpResponse response = exchange.getResponse();
360 HttpConversation conversation = exchange.getConversation();
361 ResponseNotifier notifier = connection.getDestination().getResponseNotifier();
362 notifier.notifyFailure(conversation.getResponseListeners(), response, failure);
363 LOG.debug("Failed {} {}", response, failure);
364
365 Result result = completion.getReference();
366 if (result != null)
367 {
368 connection.complete(exchange, false);
369
370 notifier.notifyComplete(conversation.getResponseListeners(), result);
371 }
372
373 return true;
374 }
375
376 @Override
377 public boolean earlyEOF()
378 {
379 failAndClose(new EOFException());
380 return false;
381 }
382
383 private void failAndClose(Throwable failure)
384 {
385 fail(failure);
386 connection.close();
387 }
388
389 @Override
390 public void badMessage(int status, String reason)
391 {
392 HttpExchange exchange = connection.getExchange();
393 HttpResponse response = exchange.getResponse();
394 response.status(status).reason(reason);
395 failAndClose(new HttpResponseException("HTTP protocol violation: bad response", response));
396 }
397
398 public void idleTimeout()
399 {
400
401
402 fail(new TimeoutException());
403 }
404
405 public boolean abort(Throwable cause)
406 {
407 return fail(cause);
408 }
409
410 private boolean updateState(State from, State to)
411 {
412 boolean updated = state.compareAndSet(from, to);
413 if (!updated)
414 LOG.debug("State update failed: {} -> {}: {}", from, to, state.get());
415 return updated;
416 }
417
418 private enum State
419 {
420 IDLE, RECEIVE, FAILURE
421 }
422 }