1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19 package org.eclipse.jetty.proxy;
20
21 import java.io.ByteArrayOutputStream;
22 import java.io.IOException;
23 import java.io.OutputStream;
24 import java.nio.ByteBuffer;
25 import java.util.ArrayDeque;
26 import java.util.ArrayList;
27 import java.util.List;
28 import java.util.Queue;
29 import java.util.concurrent.TimeUnit;
30 import java.util.zip.GZIPOutputStream;
31
32 import javax.servlet.AsyncContext;
33 import javax.servlet.ReadListener;
34 import javax.servlet.ServletConfig;
35 import javax.servlet.ServletException;
36 import javax.servlet.ServletInputStream;
37 import javax.servlet.ServletOutputStream;
38 import javax.servlet.WriteListener;
39 import javax.servlet.http.HttpServletRequest;
40 import javax.servlet.http.HttpServletResponse;
41
42 import org.eclipse.jetty.client.ContentDecoder;
43 import org.eclipse.jetty.client.GZIPContentDecoder;
44 import org.eclipse.jetty.client.api.ContentProvider;
45 import org.eclipse.jetty.client.api.Request;
46 import org.eclipse.jetty.client.api.Response;
47 import org.eclipse.jetty.client.api.Result;
48 import org.eclipse.jetty.client.util.DeferredContentProvider;
49 import org.eclipse.jetty.http.HttpHeader;
50 import org.eclipse.jetty.http.HttpVersion;
51 import org.eclipse.jetty.io.RuntimeIOException;
52 import org.eclipse.jetty.util.BufferUtil;
53 import org.eclipse.jetty.util.Callback;
54 import org.eclipse.jetty.util.CountingCallback;
55 import org.eclipse.jetty.util.IteratingCallback;
56 import org.eclipse.jetty.util.component.Destroyable;
57
58
59
60
61
62
63
64
65
66
67 public class AsyncMiddleManServlet extends AbstractProxyServlet
68 {
69 private static final String PROXY_REQUEST_COMMITTED = AsyncMiddleManServlet.class.getName() + ".proxyRequestCommitted";
70 private static final String CLIENT_TRANSFORMER = AsyncMiddleManServlet.class.getName() + ".clientTransformer";
71 private static final String SERVER_TRANSFORMER = AsyncMiddleManServlet.class.getName() + ".serverTransformer";
72
73 @Override
74 protected void service(HttpServletRequest clientRequest, HttpServletResponse proxyResponse) throws ServletException, IOException
75 {
76 String rewrittenTarget = rewriteTarget(clientRequest);
77 if (_log.isDebugEnabled())
78 {
79 StringBuffer target = clientRequest.getRequestURL();
80 if (clientRequest.getQueryString() != null)
81 target.append("?").append(clientRequest.getQueryString());
82 _log.debug("{} rewriting: {} -> {}", getRequestId(clientRequest), target, rewrittenTarget);
83 }
84 if (rewrittenTarget == null)
85 {
86 onProxyRewriteFailed(clientRequest, proxyResponse);
87 return;
88 }
89
90 final Request proxyRequest = getHttpClient().newRequest(rewrittenTarget)
91 .method(clientRequest.getMethod())
92 .version(HttpVersion.fromString(clientRequest.getProtocol()));
93
94 boolean hasContent = hasContent(clientRequest);
95
96 copyRequestHeaders(clientRequest, proxyRequest);
97
98 addProxyHeaders(clientRequest, proxyRequest);
99
100 final AsyncContext asyncContext = clientRequest.startAsync();
101
102 asyncContext.setTimeout(0);
103 proxyRequest.timeout(getTimeout(), TimeUnit.MILLISECONDS);
104
105
106
107
108 if (hasContent)
109 proxyRequest.content(newProxyContentProvider(clientRequest, proxyResponse, proxyRequest));
110 else
111 sendProxyRequest(clientRequest, proxyResponse, proxyRequest);
112 }
113
114 protected ContentProvider newProxyContentProvider(final HttpServletRequest clientRequest, HttpServletResponse proxyResponse, Request proxyRequest) throws IOException
115 {
116 ServletInputStream input = clientRequest.getInputStream();
117 DeferredContentProvider provider = new DeferredContentProvider()
118 {
119 @Override
120 public boolean offer(ByteBuffer buffer, Callback callback)
121 {
122 if (_log.isDebugEnabled())
123 _log.debug("{} proxying content to upstream: {} bytes", getRequestId(clientRequest), buffer.remaining());
124 return super.offer(buffer, callback);
125 }
126 };
127 input.setReadListener(newProxyReadListener(clientRequest, proxyResponse, proxyRequest, provider));
128 return provider;
129 }
130
131 protected ReadListener newProxyReadListener(HttpServletRequest clientRequest, HttpServletResponse proxyResponse, Request proxyRequest, DeferredContentProvider provider)
132 {
133 return new ProxyReader(clientRequest, proxyResponse, proxyRequest, provider);
134 }
135
136 protected ProxyWriter newProxyWriteListener(HttpServletRequest clientRequest, Response proxyResponse)
137 {
138 return new ProxyWriter(clientRequest, proxyResponse);
139 }
140
141 protected Response.CompleteListener newProxyResponseListener(HttpServletRequest clientRequest, HttpServletResponse proxyResponse)
142 {
143 return new ProxyResponseListener(clientRequest, proxyResponse);
144 }
145
146 protected ContentTransformer newClientRequestContentTransformer(HttpServletRequest clientRequest, Request proxyRequest)
147 {
148 return ContentTransformer.IDENTITY;
149 }
150
151 protected ContentTransformer newServerResponseContentTransformer(HttpServletRequest clientRequest, HttpServletResponse proxyResponse, Response serverResponse)
152 {
153 return ContentTransformer.IDENTITY;
154 }
155
156 private void transform(ContentTransformer transformer, ByteBuffer input, boolean finished, List<ByteBuffer> output) throws IOException
157 {
158 try
159 {
160 transformer.transform(input, finished, output);
161 }
162 catch (Throwable x)
163 {
164 _log.info("Exception while transforming " + transformer, x);
165 throw x;
166 }
167 }
168
169 int readClientRequestContent(ServletInputStream input, byte[] buffer) throws IOException
170 {
171 return input.read(buffer);
172 }
173
174 void writeProxyResponseContent(ServletOutputStream output, ByteBuffer content) throws IOException
175 {
176 write(output, content);
177 }
178
179 private static void write(OutputStream output, ByteBuffer content) throws IOException
180 {
181 int length = content.remaining();
182 int offset = 0;
183 byte[] buffer;
184 if (content.hasArray())
185 {
186 offset = content.arrayOffset();
187 buffer = content.array();
188 }
189 else
190 {
191 buffer = new byte[length];
192 content.get(buffer);
193 }
194 output.write(buffer, offset, length);
195 }
196
197 private void cleanup(HttpServletRequest clientRequest)
198 {
199 ContentTransformer clientTransformer = (ContentTransformer)clientRequest.getAttribute(CLIENT_TRANSFORMER);
200 if (clientTransformer instanceof Destroyable)
201 ((Destroyable)clientTransformer).destroy();
202 ContentTransformer serverTransformer = (ContentTransformer)clientRequest.getAttribute(SERVER_TRANSFORMER);
203 if (serverTransformer instanceof Destroyable)
204 ((Destroyable)serverTransformer).destroy();
205 }
206
207
208
209
210
211
212 public static class Transparent extends ProxyServlet
213 {
214 private final TransparentDelegate delegate = new TransparentDelegate(this);
215
216 @Override
217 public void init(ServletConfig config) throws ServletException
218 {
219 super.init(config);
220 delegate.init(config);
221 }
222
223 @Override
224 protected String rewriteTarget(HttpServletRequest request)
225 {
226 return delegate.rewriteTarget(request);
227 }
228 }
229
230 protected class ProxyReader extends IteratingCallback implements ReadListener
231 {
232 private final byte[] buffer = new byte[getHttpClient().getRequestBufferSize()];
233 private final List<ByteBuffer> buffers = new ArrayList<>();
234 private final HttpServletRequest clientRequest;
235 private final HttpServletResponse proxyResponse;
236 private final Request proxyRequest;
237 private final DeferredContentProvider provider;
238 private final int contentLength;
239 private int length;
240
241 protected ProxyReader(HttpServletRequest clientRequest, HttpServletResponse proxyResponse, Request proxyRequest, DeferredContentProvider provider)
242 {
243 this.clientRequest = clientRequest;
244 this.proxyResponse = proxyResponse;
245 this.proxyRequest = proxyRequest;
246 this.provider = provider;
247 this.contentLength = clientRequest.getContentLength();
248 }
249
250 @Override
251 public void onDataAvailable() throws IOException
252 {
253 iterate();
254 }
255
256 @Override
257 public void onAllDataRead() throws IOException
258 {
259 if (!provider.isClosed())
260 {
261 process(BufferUtil.EMPTY_BUFFER, new Adapter()
262 {
263 @Override
264 public void failed(Throwable x)
265 {
266 onError(x);
267 }
268 }, true);
269 }
270
271 if (_log.isDebugEnabled())
272 _log.debug("{} proxying content to upstream completed", getRequestId(clientRequest));
273 }
274
275 @Override
276 public void onError(Throwable t)
277 {
278 cleanup(clientRequest);
279 onClientRequestFailure(clientRequest, proxyRequest, proxyResponse, t);
280 }
281
282 @Override
283 protected Action process() throws Exception
284 {
285 ServletInputStream input = clientRequest.getInputStream();
286 while (input.isReady() && !input.isFinished())
287 {
288 int read = readClientRequestContent(input, buffer);
289 if (_log.isDebugEnabled())
290 _log.debug("{} asynchronous read {} bytes on {}", getRequestId(clientRequest), read, input);
291
292 if (contentLength > 0 && read > 0)
293 length += read;
294
295 ByteBuffer content = read > 0 ? ByteBuffer.wrap(buffer, 0, read) : BufferUtil.EMPTY_BUFFER;
296 boolean finished = read < 0 || length == contentLength;
297 process(content, this, finished);
298
299 if (read > 0)
300 return Action.SCHEDULED;
301 }
302
303 if (input.isFinished())
304 {
305 if (_log.isDebugEnabled())
306 _log.debug("{} asynchronous read complete on {}", getRequestId(clientRequest), input);
307 return Action.SUCCEEDED;
308 }
309 else
310 {
311 if (_log.isDebugEnabled())
312 _log.debug("{} asynchronous read pending on {}", getRequestId(clientRequest), input);
313 return Action.IDLE;
314 }
315 }
316
317 private void process(ByteBuffer content, Callback callback, boolean finished) throws IOException
318 {
319 ContentTransformer transformer = (ContentTransformer)clientRequest.getAttribute(CLIENT_TRANSFORMER);
320 if (transformer == null)
321 {
322 transformer = newClientRequestContentTransformer(clientRequest, proxyRequest);
323 clientRequest.setAttribute(CLIENT_TRANSFORMER, transformer);
324 }
325
326 boolean committed = clientRequest.getAttribute(PROXY_REQUEST_COMMITTED) != null;
327
328 int contentBytes = content.remaining();
329
330
331 if (contentBytes == 0 && !finished)
332 {
333 callback.succeeded();
334 return;
335 }
336
337 transform(transformer, content, finished, buffers);
338
339 int newContentBytes = 0;
340 int size = buffers.size();
341 if (size > 0)
342 {
343 CountingCallback counter = new CountingCallback(callback, size);
344 for (int i = 0; i < size; ++i)
345 {
346 ByteBuffer buffer = buffers.get(i);
347 newContentBytes += buffer.remaining();
348 provider.offer(buffer, counter);
349 }
350 buffers.clear();
351 }
352
353 if (finished)
354 provider.close();
355
356 if (_log.isDebugEnabled())
357 _log.debug("{} upstream content transformation {} -> {} bytes", getRequestId(clientRequest), contentBytes, newContentBytes);
358
359 if (!committed && (size > 0 || finished))
360 {
361 proxyRequest.header(HttpHeader.CONTENT_LENGTH, null);
362 clientRequest.setAttribute(PROXY_REQUEST_COMMITTED, true);
363 sendProxyRequest(clientRequest, proxyResponse, proxyRequest);
364 }
365
366 if (size == 0)
367 callback.succeeded();
368 }
369
370 @Override
371 protected void onCompleteFailure(Throwable x)
372 {
373 onError(x);
374 }
375 }
376
377 protected class ProxyResponseListener extends Response.Listener.Adapter implements Callback
378 {
379 private final String WRITE_LISTENER_ATTRIBUTE = AsyncMiddleManServlet.class.getName() + ".writeListener";
380
381 private final Callback complete = new CountingCallback(this, 2);
382 private final List<ByteBuffer> buffers = new ArrayList<>();
383 private final HttpServletRequest clientRequest;
384 private final HttpServletResponse proxyResponse;
385 private boolean hasContent;
386 private long contentLength;
387 private long length;
388 private Response response;
389
390 protected ProxyResponseListener(HttpServletRequest clientRequest, HttpServletResponse proxyResponse)
391 {
392 this.clientRequest = clientRequest;
393 this.proxyResponse = proxyResponse;
394 }
395
396 @Override
397 public void onBegin(Response serverResponse)
398 {
399 proxyResponse.setStatus(serverResponse.getStatus());
400 }
401
402 @Override
403 public void onHeaders(Response serverResponse)
404 {
405 contentLength = serverResponse.getHeaders().getLongField(HttpHeader.CONTENT_LENGTH.asString());
406 onServerResponseHeaders(clientRequest, proxyResponse, serverResponse);
407 }
408
409 @Override
410 public void onContent(final Response serverResponse, ByteBuffer content, final Callback callback)
411 {
412 try
413 {
414 int contentBytes = content.remaining();
415 if (_log.isDebugEnabled())
416 _log.debug("{} received server content: {} bytes", getRequestId(clientRequest), contentBytes);
417
418 hasContent = true;
419
420 ProxyWriter proxyWriter = (ProxyWriter)clientRequest.getAttribute(WRITE_LISTENER_ATTRIBUTE);
421 boolean committed = proxyWriter != null;
422 if (proxyWriter == null)
423 {
424 proxyWriter = newProxyWriteListener(clientRequest, serverResponse);
425 clientRequest.setAttribute(WRITE_LISTENER_ATTRIBUTE, proxyWriter);
426 }
427
428 ContentTransformer transformer = (ContentTransformer)clientRequest.getAttribute(SERVER_TRANSFORMER);
429 if (transformer == null)
430 {
431 transformer = newServerResponseContentTransformer(clientRequest, proxyResponse, serverResponse);
432 clientRequest.setAttribute(SERVER_TRANSFORMER, transformer);
433 }
434
435 length += contentBytes;
436
437 boolean finished = contentLength >= 0 && length == contentLength;
438 transform(transformer, content, finished, buffers);
439
440 int newContentBytes = 0;
441 int size = buffers.size();
442 if (size > 0)
443 {
444 Callback counter = size == 1 ? callback : new CountingCallback(callback, size);
445 for (int i = 0; i < size; ++i)
446 {
447 ByteBuffer buffer = buffers.get(i);
448 newContentBytes += buffer.remaining();
449 proxyWriter.offer(buffer, counter);
450 }
451 buffers.clear();
452 }
453 else
454 {
455 proxyWriter.offer(BufferUtil.EMPTY_BUFFER, callback);
456 }
457 if (finished)
458 proxyWriter.offer(BufferUtil.EMPTY_BUFFER, complete);
459
460 if (_log.isDebugEnabled())
461 _log.debug("{} downstream content transformation {} -> {} bytes", getRequestId(clientRequest), contentBytes, newContentBytes);
462
463 if (committed)
464 {
465 proxyWriter.onWritePossible();
466 }
467 else
468 {
469 if (contentLength >= 0)
470 proxyResponse.setContentLength(-1);
471
472
473
474
475
476
477
478
479 proxyResponse.getOutputStream().setWriteListener(proxyWriter);
480 }
481 }
482 catch (Throwable x)
483 {
484 callback.failed(x);
485 }
486 }
487
488 @Override
489 public void onSuccess(final Response serverResponse)
490 {
491 try
492 {
493 if (hasContent)
494 {
495
496
497 if (contentLength < 0)
498 {
499 ProxyWriter proxyWriter = (ProxyWriter)clientRequest.getAttribute(WRITE_LISTENER_ATTRIBUTE);
500 ContentTransformer transformer = (ContentTransformer)clientRequest.getAttribute(SERVER_TRANSFORMER);
501
502 transform(transformer, BufferUtil.EMPTY_BUFFER, true, buffers);
503
504 long newContentBytes = 0;
505 int size = buffers.size();
506 if (size > 0)
507 {
508 Callback callback = size == 1 ? complete : new CountingCallback(complete, size);
509 for (int i = 0; i < size; ++i)
510 {
511 ByteBuffer buffer = buffers.get(i);
512 newContentBytes += buffer.remaining();
513 proxyWriter.offer(buffer, callback);
514 }
515 buffers.clear();
516 }
517 else
518 {
519 proxyWriter.offer(BufferUtil.EMPTY_BUFFER, complete);
520 }
521
522 if (_log.isDebugEnabled())
523 _log.debug("{} downstream content transformation to {} bytes", getRequestId(clientRequest), newContentBytes);
524
525 proxyWriter.onWritePossible();
526 }
527 }
528 else
529 {
530 complete.succeeded();
531 }
532 }
533 catch (Throwable x)
534 {
535 complete.failed(x);
536 }
537 }
538
539 @Override
540 public void onComplete(Result result)
541 {
542 response = result.getResponse();
543 if (result.isSucceeded())
544 complete.succeeded();
545 else
546 complete.failed(result.getFailure());
547 }
548
549 @Override
550 public void succeeded()
551 {
552 cleanup(clientRequest);
553 onProxyResponseSuccess(clientRequest, proxyResponse, response);
554 }
555
556 @Override
557 public void failed(Throwable failure)
558 {
559 cleanup(clientRequest);
560 onProxyResponseFailure(clientRequest, proxyResponse, response, failure);
561 }
562 }
563
564 protected class ProxyWriter implements WriteListener
565 {
566 private final Queue<DeferredContentProvider.Chunk> chunks = new ArrayDeque<>();
567 private final HttpServletRequest clientRequest;
568 private final Response serverResponse;
569 private DeferredContentProvider.Chunk chunk;
570 private boolean writePending;
571
572 protected ProxyWriter(HttpServletRequest clientRequest, Response serverResponse)
573 {
574 this.clientRequest = clientRequest;
575 this.serverResponse = serverResponse;
576 }
577
578 public boolean offer(ByteBuffer content, Callback callback)
579 {
580 if (_log.isDebugEnabled())
581 _log.debug("{} proxying content to downstream: {} bytes {}", getRequestId(clientRequest), content.remaining(), callback);
582 return chunks.offer(new DeferredContentProvider.Chunk(content, callback));
583 }
584
585 @Override
586 public void onWritePossible() throws IOException
587 {
588 ServletOutputStream output = clientRequest.getAsyncContext().getResponse().getOutputStream();
589
590
591 if (writePending)
592 {
593 if (_log.isDebugEnabled())
594 _log.debug("{} pending async write complete of {} on {}", getRequestId(clientRequest), chunk, output);
595 writePending = false;
596 if (succeed(chunk.callback))
597 return;
598 }
599
600 int length = 0;
601 DeferredContentProvider.Chunk chunk = null;
602 while (output.isReady())
603 {
604 if (chunk != null)
605 {
606 if (_log.isDebugEnabled())
607 _log.debug("{} async write complete of {} ({} bytes) on {}", getRequestId(clientRequest), chunk, length, output);
608 if (succeed(chunk.callback))
609 return;
610 }
611
612 this.chunk = chunk = chunks.poll();
613 if (chunk == null)
614 return;
615
616 length = chunk.buffer.remaining();
617 if (length > 0)
618 writeProxyResponseContent(output, chunk.buffer);
619 }
620
621 if (_log.isDebugEnabled())
622 _log.debug("{} async write pending of {} ({} bytes) on {}", getRequestId(clientRequest), chunk, length, output);
623 writePending = true;
624 }
625
626 private boolean succeed(Callback callback)
627 {
628
629
630
631
632
633
634
635
636
637
638 callback.succeeded();
639 return writePending;
640 }
641
642 @Override
643 public void onError(Throwable failure)
644 {
645 DeferredContentProvider.Chunk chunk = this.chunk;
646 if (chunk != null)
647 chunk.callback.failed(failure);
648 else
649 serverResponse.abort(failure);
650 }
651 }
652
653
654
655
656
657
658
659
660
661
662 public interface ContentTransformer
663 {
664
665
666
667 public static final ContentTransformer IDENTITY = new IdentityContentTransformer();
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715 public void transform(ByteBuffer input, boolean finished, List<ByteBuffer> output) throws IOException;
716 }
717
718 private static class IdentityContentTransformer implements ContentTransformer
719 {
720 @Override
721 public void transform(ByteBuffer input, boolean finished, List<ByteBuffer> output)
722 {
723 output.add(input);
724 }
725 }
726
727 public static class GZIPContentTransformer implements ContentTransformer
728 {
729 private final List<ByteBuffer> buffers = new ArrayList<>(2);
730 private final ContentDecoder decoder = new GZIPContentDecoder();
731 private final ContentTransformer transformer;
732 private final ByteArrayOutputStream out;
733 private final GZIPOutputStream gzipOut;
734
735 public GZIPContentTransformer(ContentTransformer transformer)
736 {
737 try
738 {
739 this.transformer = transformer;
740 this.out = new ByteArrayOutputStream();
741 this.gzipOut = new GZIPOutputStream(out);
742 }
743 catch (IOException x)
744 {
745 throw new RuntimeIOException(x);
746 }
747 }
748
749 @Override
750 public void transform(ByteBuffer input, boolean finished, List<ByteBuffer> output) throws IOException
751 {
752 if (!input.hasRemaining())
753 {
754 if (finished)
755 transformer.transform(input, true, buffers);
756 }
757 else
758 {
759 while (input.hasRemaining())
760 {
761 ByteBuffer decoded = decoder.decode(input);
762 if (decoded.hasRemaining())
763 transformer.transform(decoded, finished && !input.hasRemaining(), buffers);
764 }
765 }
766
767 if (!buffers.isEmpty() || finished)
768 {
769 ByteBuffer result = gzip(buffers, finished);
770 buffers.clear();
771 output.add(result);
772 }
773 }
774
775 private ByteBuffer gzip(List<ByteBuffer> buffers, boolean finished) throws IOException
776 {
777 for (ByteBuffer buffer : buffers)
778 write(gzipOut, buffer);
779 if (finished)
780 gzipOut.close();
781 byte[] gzipBytes = out.toByteArray();
782 out.reset();
783 return ByteBuffer.wrap(gzipBytes);
784 }
785 }
786 }