1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19 package org.eclipse.jetty.fcgi.client.http;
20
21 import java.io.EOFException;
22 import java.nio.ByteBuffer;
23 import java.nio.channels.AsynchronousCloseException;
24 import java.util.LinkedList;
25 import java.util.Map;
26 import java.util.concurrent.ConcurrentHashMap;
27 import java.util.concurrent.TimeoutException;
28 import java.util.concurrent.atomic.AtomicBoolean;
29
30 import org.eclipse.jetty.client.HttpClient;
31 import org.eclipse.jetty.client.HttpConnection;
32 import org.eclipse.jetty.client.HttpDestination;
33 import org.eclipse.jetty.client.HttpExchange;
34 import org.eclipse.jetty.client.api.Connection;
35 import org.eclipse.jetty.client.api.Request;
36 import org.eclipse.jetty.client.api.Response;
37 import org.eclipse.jetty.fcgi.FCGI;
38 import org.eclipse.jetty.fcgi.generator.Flusher;
39 import org.eclipse.jetty.fcgi.parser.ClientParser;
40 import org.eclipse.jetty.http.HttpField;
41 import org.eclipse.jetty.http.HttpFields;
42 import org.eclipse.jetty.http.HttpHeader;
43 import org.eclipse.jetty.http.HttpHeaderValue;
44 import org.eclipse.jetty.io.AbstractConnection;
45 import org.eclipse.jetty.io.ByteBufferPool;
46 import org.eclipse.jetty.io.EndPoint;
47 import org.eclipse.jetty.util.BufferUtil;
48 import org.eclipse.jetty.util.CompletableCallback;
49 import org.eclipse.jetty.util.Promise;
50 import org.eclipse.jetty.util.log.Log;
51 import org.eclipse.jetty.util.log.Logger;
52
53 public class HttpConnectionOverFCGI extends AbstractConnection implements Connection
54 {
55 private static final Logger LOG = Log.getLogger(HttpConnectionOverFCGI.class);
56
57 private final LinkedList<Integer> requests = new LinkedList<>();
58 private final Map<Integer, HttpChannelOverFCGI> channels = new ConcurrentHashMap<>();
59 private final AtomicBoolean closed = new AtomicBoolean();
60 private final HttpDestination destination;
61 private final Promise<Connection> promise;
62 private final boolean multiplexed;
63 private final Flusher flusher;
64 private final Delegate delegate;
65 private final ClientParser parser;
66 private ByteBuffer buffer;
67
68 public HttpConnectionOverFCGI(EndPoint endPoint, HttpDestination destination, Promise<Connection> promise, boolean multiplexed)
69 {
70 super(endPoint, destination.getHttpClient().getExecutor());
71 this.destination = destination;
72 this.promise = promise;
73 this.multiplexed = multiplexed;
74 this.flusher = new Flusher(endPoint);
75 this.delegate = new Delegate(destination);
76 this.parser = new ClientParser(new ResponseListener());
77 requests.addLast(0);
78 }
79
80 public HttpDestination getHttpDestination()
81 {
82 return destination;
83 }
84
85 @Override
86 public void send(Request request, Response.CompleteListener listener)
87 {
88 delegate.send(request, listener);
89 }
90
91 protected void send(HttpExchange exchange)
92 {
93 delegate.send(exchange);
94 }
95
96 @Override
97 public void onOpen()
98 {
99 super.onOpen();
100 fillInterested();
101 promise.succeeded(this);
102 }
103
104 @Override
105 public void onFillable()
106 {
107 buffer = acquireBuffer();
108 process(buffer);
109 }
110
111 private ByteBuffer acquireBuffer()
112 {
113 HttpClient client = destination.getHttpClient();
114 ByteBufferPool bufferPool = client.getByteBufferPool();
115 return bufferPool.acquire(client.getResponseBufferSize(), true);
116 }
117
118 private void releaseBuffer(ByteBuffer buffer)
119 {
120 assert this.buffer == buffer;
121 HttpClient client = destination.getHttpClient();
122 ByteBufferPool bufferPool = client.getByteBufferPool();
123 bufferPool.release(buffer);
124 this.buffer = null;
125 }
126
127 private void process(ByteBuffer buffer)
128 {
129 try
130 {
131 EndPoint endPoint = getEndPoint();
132 boolean looping = false;
133 while (true)
134 {
135 if (!looping && parse(buffer))
136 return;
137
138 int read = endPoint.fill(buffer);
139 if (LOG.isDebugEnabled())
140 LOG.debug("Read {} bytes from {}", read, endPoint);
141
142 if (read > 0)
143 {
144 if (parse(buffer))
145 return;
146 }
147 else if (read == 0)
148 {
149 releaseBuffer(buffer);
150 fillInterested();
151 return;
152 }
153 else
154 {
155 releaseBuffer(buffer);
156 shutdown();
157 return;
158 }
159
160 looping = true;
161 }
162 }
163 catch (Exception x)
164 {
165 if (LOG.isDebugEnabled())
166 LOG.debug(x);
167 releaseBuffer(buffer);
168 close(x);
169 }
170 }
171
172 private boolean parse(ByteBuffer buffer)
173 {
174 return parser.parse(buffer);
175 }
176
177 private void shutdown()
178 {
179
180
181 if (channels.isEmpty())
182 close();
183 else
184 failAndClose(new EOFException(String.valueOf(getEndPoint())));
185 }
186
187 @Override
188 protected boolean onReadTimeout()
189 {
190 close(new TimeoutException());
191 return false;
192 }
193
194 protected void release(HttpChannelOverFCGI channel)
195 {
196 channels.remove(channel.getRequest());
197 destination.release(this);
198 }
199
200 @Override
201 public void close()
202 {
203 close(new AsynchronousCloseException());
204 }
205
206 protected void close(Throwable failure)
207 {
208 if (closed.compareAndSet(false, true))
209 {
210
211
212 getHttpDestination().close(this);
213 getEndPoint().shutdownOutput();
214 if (LOG.isDebugEnabled())
215 LOG.debug("{} oshut", this);
216 getEndPoint().close();
217 if (LOG.isDebugEnabled())
218 LOG.debug("{} closed", this);
219
220 abort(failure);
221 }
222 }
223
224 protected boolean closeByHTTP(HttpFields fields)
225 {
226 if (multiplexed)
227 return false;
228 if (!fields.contains(HttpHeader.CONNECTION, HttpHeaderValue.CLOSE.asString()))
229 return false;
230 close();
231 return true;
232 }
233
234 protected void abort(Throwable failure)
235 {
236 for (HttpChannelOverFCGI channel : channels.values())
237 {
238 HttpExchange exchange = channel.getHttpExchange();
239 if (exchange != null)
240 exchange.getRequest().abort(failure);
241 }
242 channels.clear();
243 }
244
245 private void failAndClose(Throwable failure)
246 {
247 boolean result = false;
248 for (HttpChannelOverFCGI channel : channels.values())
249 result |= channel.responseFailure(failure);
250 if (result)
251 close(failure);
252 }
253
254 private int acquireRequest()
255 {
256 synchronized (requests)
257 {
258 int last = requests.getLast();
259 int request = last + 1;
260 requests.addLast(request);
261 return request;
262 }
263 }
264
265 private void releaseRequest(int request)
266 {
267 synchronized (requests)
268 {
269 requests.removeFirstOccurrence(request);
270 }
271 }
272
273 @Override
274 public String toString()
275 {
276 return String.format("%s@%h(l:%s <-> r:%s)",
277 getClass().getSimpleName(),
278 this,
279 getEndPoint().getLocalAddress(),
280 getEndPoint().getRemoteAddress());
281 }
282
283 private class Delegate extends HttpConnection
284 {
285 private Delegate(HttpDestination destination)
286 {
287 super(destination);
288 }
289
290 @Override
291 protected void send(HttpExchange exchange)
292 {
293 Request request = exchange.getRequest();
294 normalizeRequest(request);
295
296
297 int id = acquireRequest();
298 HttpChannelOverFCGI channel = new HttpChannelOverFCGI(HttpConnectionOverFCGI.this, flusher, id, request.getIdleTimeout());
299 channels.put(id, channel);
300 if (channel.associate(exchange))
301 channel.send();
302 else
303 channel.release();
304 }
305
306 @Override
307 public void close()
308 {
309 HttpConnectionOverFCGI.this.close();
310 }
311
312 @Override
313 public String toString()
314 {
315 return HttpConnectionOverFCGI.this.toString();
316 }
317 }
318
319 private class ResponseListener implements ClientParser.Listener
320 {
321 @Override
322 public void onBegin(int request, int code, String reason)
323 {
324 HttpChannelOverFCGI channel = channels.get(request);
325 if (channel != null)
326 channel.responseBegin(code, reason);
327 else
328 noChannel(request);
329 }
330
331 @Override
332 public void onHeader(int request, HttpField field)
333 {
334 HttpChannelOverFCGI channel = channels.get(request);
335 if (channel != null)
336 channel.responseHeader(field);
337 else
338 noChannel(request);
339 }
340
341 @Override
342 public void onHeaders(int request)
343 {
344 HttpChannelOverFCGI channel = channels.get(request);
345 if (channel != null)
346 channel.responseHeaders();
347 else
348 noChannel(request);
349 }
350
351 @Override
352 public boolean onContent(int request, FCGI.StreamType stream, ByteBuffer buffer)
353 {
354 switch (stream)
355 {
356 case STD_OUT:
357 {
358 HttpChannelOverFCGI channel = channels.get(request);
359 if (channel != null)
360 {
361 CompletableCallback callback = new CompletableCallback()
362 {
363 @Override
364 public void resume()
365 {
366 if (LOG.isDebugEnabled())
367 LOG.debug("Content consumed asynchronously, resuming processing");
368 process(HttpConnectionOverFCGI.this.buffer);
369 }
370
371 @Override
372 public void abort(Throwable x)
373 {
374 close(x);
375 }
376 };
377 if (!channel.content(buffer, callback))
378 return true;
379 return callback.tryComplete();
380 }
381 else
382 {
383 noChannel(request);
384 }
385 break;
386 }
387 case STD_ERR:
388 {
389 LOG.info(BufferUtil.toUTF8String(buffer));
390 break;
391 }
392 default:
393 {
394 throw new IllegalArgumentException();
395 }
396 }
397 return false;
398 }
399
400 @Override
401 public void onEnd(int request)
402 {
403 HttpChannelOverFCGI channel = channels.get(request);
404 if (channel != null)
405 {
406 if (channel.responseSuccess())
407 releaseRequest(request);
408 }
409 else
410 {
411 noChannel(request);
412 }
413 }
414
415 @Override
416 public void onFailure(int request, Throwable failure)
417 {
418 HttpChannelOverFCGI channel = channels.get(request);
419 if (channel != null)
420 {
421 if (channel.responseFailure(failure))
422 releaseRequest(request);
423 }
424 else
425 {
426 noChannel(request);
427 }
428 }
429
430 private void noChannel(int request)
431 {
432 if (LOG.isDebugEnabled())
433 LOG.debug("Channel not found for request {}", request);
434 }
435 }
436 }