1 //
2 // ========================================================================
3 // Copyright (c) 1995-2016 Mort Bay Consulting Pty. Ltd.
4 // ------------------------------------------------------------------------
5 // All rights reserved. This program and the accompanying materials
6 // are made available under the terms of the Eclipse Public License v1.0
7 // and Apache License v2.0 which accompanies this distribution.
8 //
9 // The Eclipse Public License is available at
10 // http://www.eclipse.org/legal/epl-v10.html
11 //
12 // The Apache License v2.0 is available at
13 // http://www.opensource.org/licenses/apache2.0.php
14 //
15 // You may elect to redistribute this code under either of these licenses.
16 // ========================================================================
17 //
18
19 package org.eclipse.jetty.server;
20
21 import java.io.IOException;
22 import java.nio.ByteBuffer;
23 import java.nio.charset.StandardCharsets;
24 import java.util.concurrent.BlockingQueue;
25 import java.util.concurrent.CountDownLatch;
26 import java.util.concurrent.Executor;
27 import java.util.concurrent.LinkedBlockingQueue;
28 import java.util.concurrent.TimeUnit;
29
30 import org.eclipse.jetty.http.HttpField;
31 import org.eclipse.jetty.http.HttpParser;
32 import org.eclipse.jetty.http.HttpVersion;
33 import org.eclipse.jetty.io.ByteArrayEndPoint;
34 import org.eclipse.jetty.io.ByteBufferPool;
35 import org.eclipse.jetty.io.Connection;
36 import org.eclipse.jetty.util.BufferUtil;
37 import org.eclipse.jetty.util.ByteArrayOutputStream2;
38 import org.eclipse.jetty.util.ssl.SslContextFactory;
39 import org.eclipse.jetty.util.thread.Scheduler;
40
41 /**
42 * A local connector, mostly for testing purposes.
43 * <pre>
44 * HttpTester.Request request = HttpTester.newRequest();
45 * request.setURI("/some/resource");
46 * HttpTester.Response response =
47 * HttpTester.parseResponse(HttpTester.from(localConnector.getResponse(request.generate())));
48 * </pre>
49 *
50 */
51 public class LocalConnector extends AbstractConnector
52 {
53 private final BlockingQueue<LocalEndPoint> _connects = new LinkedBlockingQueue<>();
54
55 public LocalConnector(Server server, Executor executor, Scheduler scheduler, ByteBufferPool pool, int acceptors, ConnectionFactory... factories)
56 {
57 super(server,executor,scheduler,pool,acceptors,factories);
58 setIdleTimeout(30000);
59 }
60
61 public LocalConnector(Server server)
62 {
63 this(server, null, null, null, -1, new HttpConnectionFactory());
64 }
65
66 public LocalConnector(Server server, SslContextFactory sslContextFactory)
67 {
68 this(server, null, null, null, -1,AbstractConnectionFactory.getFactories(sslContextFactory,new HttpConnectionFactory()));
69 }
70
71 public LocalConnector(Server server, ConnectionFactory connectionFactory)
72 {
73 this(server, null, null, null, -1, connectionFactory);
74 }
75
76 public LocalConnector(Server server, ConnectionFactory connectionFactory, SslContextFactory sslContextFactory)
77 {
78 this(server, null, null, null, -1, AbstractConnectionFactory.getFactories(sslContextFactory,connectionFactory));
79 }
80
81 @Override
82 public Object getTransport()
83 {
84 return this;
85 }
86
87 /** Sends requests and get responses based on thread activity.
88 * Returns all the responses received once the thread activity has
89 * returned to the level it was before the requests.
90 * <p>
91 * This methods waits until the connection is closed or
92 * is idle for 5s before returning the responses.
93 * <p>Use {@link #getResponse(String)} for an alternative that does not wait for idle.
94 * @param requests the requests
95 * @return the responses
96 * @throws Exception if the requests fail
97 */
98 public String getResponses(String requests) throws Exception
99 {
100 return getResponses(requests, 5, TimeUnit.SECONDS);
101 }
102
103 /** Sends requests and get responses based on thread activity.
104 * Returns all the responses received once the thread activity has
105 * returned to the level it was before the requests.
106 * <p>
107 * This methods waits until the connection is closed or
108 * an idle period before returning the responses.
109 * <p>Use {@link #getResponse(String)} for an alternative that does not wait for idle.
110 * @param requests the requests
111 * @param idleFor The time the response stream must be idle for before returning
112 * @param units The units of idleFor
113 * @return the responses
114 * @throws Exception if the requests fail
115 */
116 public String getResponses(String requests,long idleFor,TimeUnit units) throws Exception
117 {
118 ByteBuffer result = getResponses(BufferUtil.toBuffer(requests,StandardCharsets.UTF_8),idleFor,units);
119 return result==null?null:BufferUtil.toString(result,StandardCharsets.UTF_8);
120 }
121
122 /** Sends requests and get's responses based on thread activity.
123 * Returns all the responses received once the thread activity has
124 * returned to the level it was before the requests.
125 * <p>
126 * This methods waits until the connection is closed or
127 * is idle for 5s before returning the responses.
128 * <p>Use {@link #getResponse(ByteBuffer)} for an alternative that does not wait for idle.
129 * @param requestsBuffer the requests
130 * @return the responses
131 * @throws Exception if the requests fail
132 */
133 public ByteBuffer getResponses(ByteBuffer requestsBuffer) throws Exception
134 {
135 return getResponses(requestsBuffer, 5, TimeUnit.SECONDS);
136 }
137
138 /** Sends requests and get's responses based on thread activity.
139 * Returns all the responses received once the thread activity has
140 * returned to the level it was before the requests.
141 * <p>
142 * This methods waits until the connection is closed or
143 * an idle period before returning the responses.
144 * @param requestsBuffer the requests
145 * @param idleFor The time the response stream must be idle for before returning
146 * @param units The units of idleFor
147 * @return the responses
148 * @throws Exception if the requests fail
149 */
150 public ByteBuffer getResponses(ByteBuffer requestsBuffer,long idleFor,TimeUnit units) throws Exception
151 {
152 if (LOG.isDebugEnabled())
153 LOG.debug("requests {}", BufferUtil.toUTF8String(requestsBuffer));
154 LocalEndPoint endp = executeRequest(requestsBuffer);
155 endp.waitUntilClosedOrIdleFor(idleFor,units);
156 ByteBuffer responses = endp.takeOutput();
157 if (endp.isOutputShutdown())
158 endp.close();
159 if (LOG.isDebugEnabled())
160 LOG.debug("responses {}", BufferUtil.toUTF8String(responses));
161 return responses;
162 }
163
164 /**
165 * Execute a request and return the EndPoint through which
166 * multiple responses can be received or more input provided.
167 * @param rawRequest the request
168 * @return the local endpoint
169 */
170 public LocalEndPoint executeRequest(String rawRequest)
171 {
172 return executeRequest(BufferUtil.toBuffer(rawRequest, StandardCharsets.UTF_8));
173 }
174
175 private LocalEndPoint executeRequest(ByteBuffer rawRequest)
176 {
177 if (!isStarted())
178 throw new IllegalStateException("!STARTED");
179 LocalEndPoint endp = new LocalEndPoint();
180 endp.addInput(rawRequest);
181 _connects.add(endp);
182 return endp;
183 }
184
185 public LocalEndPoint connect()
186 {
187 LocalEndPoint endp = new LocalEndPoint();
188 _connects.add(endp);
189 return endp;
190 }
191
192 @Override
193 protected void accept(int acceptorID) throws IOException, InterruptedException
194 {
195 if (LOG.isDebugEnabled())
196 LOG.debug("accepting {}", acceptorID);
197 LocalEndPoint endPoint = _connects.take();
198 endPoint.onOpen();
199 onEndPointOpened(endPoint);
200
201 Connection connection = getDefaultConnectionFactory().newConnection(this, endPoint);
202 endPoint.setConnection(connection);
203
204 connection.onOpen();
205 }
206
207
208 /** Get a single response using a parser to search for the end of the message.
209 * @param requestsBuffer The request to send
210 * @return ByteBuffer containing response or null.
211 * @throws Exception If there is a problem
212 */
213 public ByteBuffer getResponse(ByteBuffer requestsBuffer) throws Exception
214 {
215 return getResponse(requestsBuffer,false,10,TimeUnit.SECONDS);
216 }
217
218 /** Get a single response using a parser to search for the end of the message.
219 * @param requestBuffer The request to send
220 * @param time The time to wait
221 * @param unit The units of the wait
222 * @return ByteBuffer containing response or null.
223 * @throws Exception If there is a problem
224 */
225 public ByteBuffer getResponse(ByteBuffer requestBuffer, long time,TimeUnit unit) throws Exception
226 {
227 boolean head = BufferUtil.toString(requestBuffer).toLowerCase().startsWith("head ");
228 if (LOG.isDebugEnabled())
229 LOG.debug("requests {}", BufferUtil.toUTF8String(requestBuffer));
230 LocalEndPoint endp = executeRequest(requestBuffer);
231 return endp.waitForResponse(head,time,unit);
232 }
233
234 /** Get a single response using a parser to search for the end of the message.
235 * @param requestBuffer The request to send
236 * @param head True if the response is for a head request
237 * @param time The time to wait
238 * @param unit The units of the wait
239 * @return ByteBuffer containing response or null.
240 * @throws Exception If there is a problem
241 */
242 public ByteBuffer getResponse(ByteBuffer requestBuffer,boolean head, long time,TimeUnit unit) throws Exception
243 {
244 if (LOG.isDebugEnabled())
245 LOG.debug("requests {}", BufferUtil.toUTF8String(requestBuffer));
246 LocalEndPoint endp = executeRequest(requestBuffer);
247 return endp.waitForResponse(head,time,unit);
248 }
249
250
251 /** Get a single response using a parser to search for the end of the message.
252 * @param rawRequest The request to send
253 * @return ByteBuffer containing response or null.
254 * @throws Exception If there is a problem
255 */
256 public String getResponse(String rawRequest) throws Exception
257 {
258 return getResponse(rawRequest,false,30,TimeUnit.SECONDS);
259 }
260
261 /** Get a single response using a parser to search for the end of the message.
262 * @param rawRequest The request to send
263 * @param time The time to wait
264 * @param unit The units of the wait
265 * @return ByteBuffer containing response or null.
266 * @throws Exception If there is a problem
267 */
268 public String getResponse(String rawRequest,long time,TimeUnit unit) throws Exception
269 {
270 boolean head = rawRequest.toLowerCase().startsWith("head ");
271 ByteBuffer requestsBuffer = BufferUtil.toBuffer(rawRequest, StandardCharsets.ISO_8859_1);
272 if (LOG.isDebugEnabled())
273 LOG.debug("request {}", BufferUtil.toUTF8String(requestsBuffer));
274 LocalEndPoint endp = executeRequest(requestsBuffer);
275
276 return BufferUtil.toString(endp.waitForResponse(head,time,unit), StandardCharsets.ISO_8859_1);
277 }
278
279
280 /** Get a single response using a parser to search for the end of the message.
281 * @param rawRequest The request to send
282 * @param head True if the response is for a head request
283 * @param time The time to wait
284 * @param unit The units of the wait
285 * @return ByteBuffer containing response or null.
286 * @throws Exception If there is a problem
287 */
288 public String getResponse(String rawRequest, boolean head, long time,TimeUnit unit) throws Exception
289 {
290 ByteBuffer requestsBuffer = BufferUtil.toBuffer(rawRequest, StandardCharsets.ISO_8859_1);
291 if (LOG.isDebugEnabled())
292 LOG.debug("request {}", BufferUtil.toUTF8String(requestsBuffer));
293 LocalEndPoint endp = executeRequest(requestsBuffer);
294
295 return BufferUtil.toString(endp.waitForResponse(head,time,unit), StandardCharsets.ISO_8859_1);
296 }
297
298 /** Local EndPoint
299 */
300 public class LocalEndPoint extends ByteArrayEndPoint
301 {
302 private final CountDownLatch _closed = new CountDownLatch(1);
303 private ByteBuffer _responseData;
304
305 public LocalEndPoint()
306 {
307 super(LocalConnector.this.getScheduler(), LocalConnector.this.getIdleTimeout());
308 setGrowOutput(true);
309 }
310
311 protected void execute(Runnable task)
312 {
313 getExecutor().execute(task);
314 }
315
316 @Override
317 public void close()
318 {
319 boolean wasOpen=isOpen();
320 super.close();
321 if (wasOpen)
322 {
323 getConnection().onClose();
324 onClose();
325 }
326 }
327
328 @Override
329 public void onClose()
330 {
331 LocalConnector.this.onEndPointClosed(this);
332 super.onClose();
333 _closed.countDown();
334 }
335
336 @Override
337 public void shutdownOutput()
338 {
339 super.shutdownOutput();
340 close();
341 }
342
343 public void waitUntilClosed()
344 {
345 while (isOpen())
346 {
347 try
348 {
349 if (!_closed.await(10,TimeUnit.SECONDS))
350 break;
351 }
352 catch(Exception e)
353 {
354 LOG.warn(e);
355 }
356 }
357 }
358
359 public void waitUntilClosedOrIdleFor(long idleFor,TimeUnit units)
360 {
361 Thread.yield();
362 int size=getOutput().remaining();
363 while (isOpen())
364 {
365 try
366 {
367 if (!_closed.await(idleFor,units))
368 {
369 if (size==getOutput().remaining())
370 {
371 if (LOG.isDebugEnabled())
372 LOG.debug("idle for {} {}",idleFor,units);
373 return;
374 }
375 size=getOutput().remaining();
376 }
377 }
378 catch(Exception e)
379 {
380 LOG.warn(e);
381 }
382 }
383 }
384
385 /**
386 * Wait for a response using a parser to detect the end of message
387 * @return Buffer containing full response or null for EOF;
388 * @throws Exception if the response cannot be parsed
389 */
390 public String getResponse() throws Exception
391 {
392 return getResponse(false,30,TimeUnit.SECONDS);
393 }
394
395 /**
396 * Wait for a response using a parser to detect the end of message
397 * @param head whether the request is a HEAD request
398 * @param time the maximum time to wait
399 * @param unit the time unit of the {@code timeout} argument
400 * @return Buffer containing full response or null for EOF;
401 * @throws Exception if the response cannot be parsed
402 */
403 public String getResponse(boolean head, long time,TimeUnit unit) throws Exception
404 {
405 ByteBuffer response = waitForResponse(head,time,unit);
406 if (response!=null)
407 return BufferUtil.toString(response);
408 return null;
409 }
410
411 /**
412 * Wait for a response using a parser to detect the end of message
413 * @param head whether the request is a HEAD request
414 * @param time the maximum time to wait
415 * @param unit the time unit of the {@code timeout} argument
416 * @return Buffer containing full response or null for EOF;
417 * @throws Exception if the response cannot be parsed
418 */
419 public ByteBuffer waitForResponse(boolean head, long time,TimeUnit unit) throws Exception
420 {
421 HttpParser.ResponseHandler handler = new HttpParser.ResponseHandler()
422 {
423 @Override
424 public void parsedHeader(HttpField field)
425 {
426 }
427
428 @Override
429 public boolean messageComplete()
430 {
431 return true;
432 }
433
434 @Override
435 public boolean headerComplete()
436 {
437 return false;
438 }
439
440 @Override
441 public int getHeaderCacheSize()
442 {
443 return 0;
444 }
445
446 @Override
447 public void earlyEOF()
448 {
449 }
450
451 @Override
452 public boolean content(ByteBuffer item)
453 {
454 return false;
455 }
456
457 @Override
458 public void badMessage(int status, String reason)
459 {
460 }
461
462 @Override
463 public boolean startResponse(HttpVersion version, int status, String reason)
464 {
465 return false;
466 }
467 };
468
469 HttpParser parser = new HttpParser(handler);
470 parser.setHeadResponse(head);
471 try(ByteArrayOutputStream2 bout = new ByteArrayOutputStream2();)
472 {
473 loop: while(true)
474 {
475 // read a chunk of response
476 ByteBuffer chunk;
477 if (BufferUtil.hasContent(_responseData))
478 chunk = _responseData;
479 else
480 {
481 chunk = waitForOutput(time,unit);
482 if (BufferUtil.isEmpty(chunk) && (!isOpen() || isOutputShutdown()))
483 {
484 parser.atEOF();
485 parser.parseNext(BufferUtil.EMPTY_BUFFER);
486 break loop;
487 }
488 }
489
490 // Parse the content of this chunk
491 while (BufferUtil.hasContent(chunk))
492 {
493 int pos=chunk.position();
494 boolean complete=parser.parseNext(chunk);
495 if (chunk.position()==pos)
496 {
497 // Nothing consumed
498 if (BufferUtil.isEmpty(chunk))
499 break;
500 return null;
501 }
502
503 // Add all consumed bytes to the output stream
504 bout.write(chunk.array(),chunk.arrayOffset()+pos,chunk.position()-pos);
505
506 // If we are complete then break the outer loop
507 if (complete)
508 {
509 if (BufferUtil.hasContent(chunk))
510 _responseData=chunk;
511 break loop;
512 }
513 }
514 }
515
516 if (bout.getCount()==0 && isOutputShutdown())
517 return null;
518 return ByteBuffer.wrap(bout.getBuf(),0,bout.getCount());
519 }
520 }
521 }
522 }