1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19 package org.eclipse.jetty.client.http;
20
21 import java.io.EOFException;
22 import java.nio.ByteBuffer;
23
24 import org.eclipse.jetty.client.HttpClient;
25 import org.eclipse.jetty.client.HttpExchange;
26 import org.eclipse.jetty.client.HttpReceiver;
27 import org.eclipse.jetty.client.HttpResponse;
28 import org.eclipse.jetty.client.HttpResponseException;
29 import org.eclipse.jetty.http.HttpField;
30 import org.eclipse.jetty.http.HttpMethod;
31 import org.eclipse.jetty.http.HttpParser;
32 import org.eclipse.jetty.http.HttpVersion;
33 import org.eclipse.jetty.io.ByteBufferPool;
34 import org.eclipse.jetty.io.EndPoint;
35 import org.eclipse.jetty.util.BufferUtil;
36 import org.eclipse.jetty.util.CompletableCallback;
37
38 public class HttpReceiverOverHTTP extends HttpReceiver implements HttpParser.ResponseHandler<ByteBuffer>
39 {
40 private final HttpParser parser = new HttpParser(this);
41 private ByteBuffer buffer;
42 private boolean shutdown;
43
44 public HttpReceiverOverHTTP(HttpChannelOverHTTP channel)
45 {
46 super(channel);
47 }
48
49 @Override
50 public HttpChannelOverHTTP getHttpChannel()
51 {
52 return (HttpChannelOverHTTP)super.getHttpChannel();
53 }
54
55 private HttpConnectionOverHTTP getHttpConnection()
56 {
57 return getHttpChannel().getHttpConnection();
58 }
59
60 protected ByteBuffer getResponseBuffer()
61 {
62 return buffer;
63 }
64
65 public void receive()
66 {
67 if (buffer == null)
68 acquireBuffer();
69 process();
70 }
71
72 private void acquireBuffer()
73 {
74 HttpClient client = getHttpDestination().getHttpClient();
75 ByteBufferPool bufferPool = client.getByteBufferPool();
76 buffer = bufferPool.acquire(client.getResponseBufferSize(), true);
77 }
78
79 private void releaseBuffer()
80 {
81 if (buffer == null)
82 throw new IllegalStateException();
83 if (BufferUtil.hasContent(buffer))
84 throw new IllegalStateException();
85 HttpClient client = getHttpDestination().getHttpClient();
86 ByteBufferPool bufferPool = client.getByteBufferPool();
87 bufferPool.release(buffer);
88 buffer = null;
89 }
90
91 private void process()
92 {
93 try
94 {
95 HttpConnectionOverHTTP connection = getHttpConnection();
96 EndPoint endPoint = connection.getEndPoint();
97 while (true)
98 {
99
100 if (connection.isClosed())
101 {
102 if (LOG.isDebugEnabled())
103 LOG.debug("{} closed", connection);
104 releaseBuffer();
105 return;
106 }
107
108 if (parse())
109 return;
110
111 int read = endPoint.fill(buffer);
112 if (LOG.isDebugEnabled())
113 LOG.debug("Read {} bytes {} from {}", read, BufferUtil.toDetailString(buffer), endPoint);
114
115 if (read > 0)
116 {
117 if (parse())
118 return;
119 }
120 else if (read == 0)
121 {
122 releaseBuffer();
123 fillInterested();
124 return;
125 }
126 else
127 {
128 releaseBuffer();
129 shutdown();
130 return;
131 }
132 }
133 }
134 catch (Throwable x)
135 {
136 if (LOG.isDebugEnabled())
137 LOG.debug(x);
138 BufferUtil.clear(buffer);
139 if (buffer != null)
140 releaseBuffer();
141 failAndClose(x);
142 }
143 }
144
145
146
147
148
149
150 private boolean parse()
151 {
152
153
154 boolean handle = parser.parseNext(buffer);
155 if (LOG.isDebugEnabled())
156 LOG.debug("Parsed {}, remaining {} {}", handle, buffer.remaining(), parser);
157
158 return handle;
159 }
160
161 protected void fillInterested()
162 {
163 getHttpChannel().getHttpConnection().fillInterested();
164 }
165
166 private void shutdown()
167 {
168
169
170
171
172 shutdown = true;
173
174
175
176
177
178 parser.atEOF();
179 parser.parseNext(BufferUtil.EMPTY_BUFFER);
180 }
181
182 protected boolean isShutdown()
183 {
184 return shutdown;
185 }
186
187 @Override
188 public int getHeaderCacheSize()
189 {
190
191 return 256;
192 }
193
194 @Override
195 public boolean startResponse(HttpVersion version, int status, String reason)
196 {
197 HttpExchange exchange = getHttpExchange();
198 if (exchange == null)
199 return false;
200
201 String method = exchange.getRequest().getMethod();
202 parser.setHeadResponse(HttpMethod.HEAD.is(method) || HttpMethod.CONNECT.is(method));
203 exchange.getResponse().version(version).status(status).reason(reason);
204
205 responseBegin(exchange);
206 return false;
207 }
208
209 @Override
210 public boolean parsedHeader(HttpField field)
211 {
212 HttpExchange exchange = getHttpExchange();
213 if (exchange == null)
214 return false;
215
216 responseHeader(exchange, field);
217 return false;
218 }
219
220 @Override
221 public boolean headerComplete()
222 {
223 HttpExchange exchange = getHttpExchange();
224 if (exchange == null)
225 return false;
226
227 responseHeaders(exchange);
228 return false;
229 }
230
231 @Override
232 public boolean content(ByteBuffer buffer)
233 {
234 HttpExchange exchange = getHttpExchange();
235 if (exchange == null)
236 return false;
237
238 CompletableCallback callback = new CompletableCallback()
239 {
240 @Override
241 public void resume()
242 {
243 if (LOG.isDebugEnabled())
244 LOG.debug("Content consumed asynchronously, resuming processing");
245 process();
246 }
247
248 public void abort(Throwable x)
249 {
250 failAndClose(x);
251 }
252 };
253 responseContent(exchange, buffer, callback);
254 return callback.tryComplete();
255 }
256
257 @Override
258 public boolean messageComplete()
259 {
260 HttpExchange exchange = getHttpExchange();
261 if (exchange != null)
262 responseSuccess(exchange);
263 return false;
264 }
265
266 @Override
267 public void earlyEOF()
268 {
269 HttpExchange exchange = getHttpExchange();
270 HttpConnectionOverHTTP connection = getHttpConnection();
271 if (exchange == null)
272 connection.close();
273 else
274 failAndClose(new EOFException(String.valueOf(connection)));
275 }
276
277 @Override
278 public void badMessage(int status, String reason)
279 {
280 HttpExchange exchange = getHttpExchange();
281 if (exchange != null)
282 {
283 HttpResponse response = exchange.getResponse();
284 response.status(status).reason(reason);
285 failAndClose(new HttpResponseException("HTTP protocol violation: bad response on " + getHttpConnection(), response));
286 }
287 }
288
289 @Override
290 protected void reset()
291 {
292 super.reset();
293 parser.reset();
294 }
295
296 @Override
297 protected void dispose()
298 {
299 super.dispose();
300 parser.close();
301 }
302
303 private void failAndClose(Throwable failure)
304 {
305 if (responseFailure(failure))
306 getHttpConnection().close(failure);
307 }
308
309 @Override
310 public String toString()
311 {
312 return String.format("%s[%s]", super.toString(), parser);
313 }
314 }