Assist in capturing and handling exceptions or errors within interceptors to prevent request hanging (#84)

* ### Assist in capturing and handling exceptions or errors within interceptors to prevent request hanging, which also might address issue #75.

This may help solve #75.  In addition, when interceptors fail to filter the request or response in the chain, this can return a response in time, preventing the request from hanging.

* Release resources of the request before send the response

* Log errors during filtering the request/response, simplify the code

Author: yameizi001

src/main/java/io/vertx/httpproxy/impl/ReverseProxy.java

@@ -11,8 +11,9 @@
 package io.vertx.httpproxy.impl;
 
 import io.vertx.core.Future;
-import io.vertx.core.Promise;
 import io.vertx.core.http.*;
+import io.vertx.core.internal.logging.Logger;
+import io.vertx.core.internal.logging.LoggerFactory;
 import io.vertx.core.net.NetSocket;
 import io.vertx.httpproxy.*;
 import io.vertx.httpproxy.cache.CacheOptions;
@@ -23,6 +24,7 @@
 
 public class ReverseProxy implements HttpProxy {
 
+  private final static Logger log = LoggerFactory.getLogger(ReverseProxy.class);
   private final HttpClient client;
   private final boolean supportWebSocket;
   private BiFunction<HttpServerRequest, HttpClient, Future<HttpClientRequest>> selector = (req, client) -> Future.failedFuture("No origin available");
@@ -70,7 +72,16 @@ public void handle(HttpServerRequest request) {
 
     Proxy proxy = new Proxy(proxyRequest);
     proxy.filters = interceptors.listIterator();
-    proxy.sendRequest().compose(proxy::sendProxyResponse);
+    proxy.sendRequest()
+      .recover(throwable -> {
+        log.trace("Error in sending the request", throwable);
+        return Future.succeededFuture(proxyRequest.release().response().setStatusCode(502));
+      })
+      .compose(proxy::sendProxyResponse)
+      .recover(throwable -> {
+        log.trace("Error in sending the response", throwable);
+        return proxy.response().release().setStatusCode(502).send();
+      });
   }
 
   private void handleWebSocketUpgrade(ProxyRequest proxyRequest) {
@@ -191,27 +202,7 @@ public Future<Void> sendResponse() {
     }
 
     private Future<ProxyResponse> sendProxyRequest(ProxyRequest proxyRequest) {
-      Future<HttpClientRequest> f = resolveOrigin(proxyRequest.proxiedRequest());
-      f.onFailure(err -> {
-        // Should this be done here ? I don't think so
-        HttpServerRequest proxiedRequest = proxyRequest.proxiedRequest();
-        proxiedRequest.resume();
-        Promise<Void> promise = Promise.promise();
-        proxiedRequest.exceptionHandler(promise::tryFail);
-        proxiedRequest.endHandler(promise::tryComplete);
-        promise.future().onComplete(ar2 -> {
-          end(proxyRequest, 502);
-        });
-      });
-      return f.compose(a -> sendProxyRequest(proxyRequest, a));
-    }
-
-    private Future<ProxyResponse> sendProxyRequest(ProxyRequest proxyRequest, HttpClientRequest request) {
-      Future<ProxyResponse> fut = proxyRequest.send(request);
-      fut.onFailure(err -> {
-        proxyRequest.proxiedRequest().response().setStatusCode(502).end();
-      });
-      return fut;
+      return resolveOrigin(proxyRequest.proxiedRequest()).compose(proxyRequest::send);
     }
 
     private Future<Void> sendProxyResponse(ProxyResponse response) {

src/test/java/io/vertx/httpproxy/ProxyTest.java

@@ -110,4 +110,56 @@ public Future<Void> handleProxyResponse(ProxyContext context) {
         latch.countDown();
       });
   }
+
+  @Test
+  public void testUpstreamRefuse(TestContext ctx) {
+    SocketAddress backend = SocketAddress.inetSocketAddress(8081, "localhost");
+    startProxy(proxy -> proxy.origin(backend));
+    client = vertx.createHttpClient();
+    Async async = ctx.async();
+    client.request(HttpMethod.GET, 8080, "localhost", "/")
+      .compose(req -> req.send().compose(resp -> {
+        ctx.assertEquals(502, resp.statusCode());
+        return resp.body();
+      }))
+      .onComplete(ctx.asyncAssertSuccess(body -> async.complete()));
+  }
+
+  @Test
+  public void testFilterRequestFail(TestContext ctx) {
+    SocketAddress backend = startHttpBackend(ctx, 8081, req -> req.response().end("HOLA"));
+    startProxy(proxy -> proxy.origin(backend).addInterceptor(new ProxyInterceptor() {
+      @Override
+      public Future<ProxyResponse> handleProxyRequest(ProxyContext context) {
+        return Future.failedFuture(new RuntimeException("Some error"));
+      }
+    }));
+    client = vertx.createHttpClient();
+    Async async = ctx.async();
+    client.request(HttpMethod.GET, 8080, "localhost", "/")
+      .compose(req -> req.send().compose(resp -> {
+        ctx.assertEquals(502, resp.statusCode());
+        return resp.body();
+      }))
+      .onComplete(ctx.asyncAssertSuccess(body -> async.complete()));
+  }
+
+  @Test
+  public void testFilterResponseFail(TestContext ctx) {
+    SocketAddress backend = startHttpBackend(ctx, 8081, req -> req.response().end("HOLA"));
+    startProxy(proxy -> proxy.origin(backend).addInterceptor(new ProxyInterceptor() {
+      @Override
+      public Future<Void> handleProxyResponse(ProxyContext context) {
+        return Future.failedFuture(new RuntimeException("Some error"));
+      }
+    }));
+    client = vertx.createHttpClient();
+    Async async = ctx.async();
+    client.request(HttpMethod.GET, 8080, "localhost", "/")
+      .compose(req -> req.send().compose(resp -> {
+        ctx.assertEquals(502, resp.statusCode());
+        return resp.body();
+      }))
+      .onComplete(ctx.asyncAssertSuccess(body -> async.complete()));
+  }
 }