Merge pull request cdapio#17 from caskdata/feature/exception-handler

NETTY-4: Exception handler

Author: jwang47

src/main/java/co/cask/http/ExceptionHandler.java

@@ -0,0 +1,30 @@
+/*
+ * Copyright © 2015 Cask Data, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License"); you may not
+ * use this file except in compliance with the License. You may obtain a copy of
+ * the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+ * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+ * License for the specific language governing permissions and limitations under
+ * the License.
+ */
+
+package co.cask.http;
+
+import org.jboss.netty.handler.codec.http.HttpRequest;
+import org.jboss.netty.handler.codec.http.HttpResponseStatus;
+
+/**
+ * Handles exceptions and provides a response via the {@link HttpResponder}.
+ */
+public class ExceptionHandler {
+  public void handle(Throwable t, HttpRequest request, HttpResponder responder) {
+    String message = String.format("Exception encountered while processing request : %s", t.getMessage());
+    responder.sendString(HttpResponseStatus.INTERNAL_SERVER_ERROR, message);
+  }
+}

src/main/java/co/cask/http/HttpMethodInfo.java

@@ -39,23 +39,28 @@ class HttpMethodInfo {
   private final HttpHandler handler;
   private final boolean isChunkedRequest;
   private final ChannelBuffer requestContent;
+  private final HttpRequest request;
   private final HttpResponder responder;
   private final Object[] args;
   private final boolean isStreaming;
+  private final ExceptionHandler exceptionHandler;
 
   private BodyConsumer bodyConsumer;
 
-  HttpMethodInfo(Method method, HttpHandler handler, HttpRequest request, HttpResponder responder, Object[] args) {
+  HttpMethodInfo(Method method, HttpHandler handler, HttpRequest request, HttpResponder responder, Object[] args,
+                 ExceptionHandler exceptionHandler) {
     this.method = method;
     this.handler = handler;
     this.isChunkedRequest = request.isChunked();
     this.requestContent = request.getContent();
-    this.responder = responder;
     this.isStreaming = BodyConsumer.class.isAssignableFrom(method.getReturnType());
+    this.request = rewriteRequest(request, isStreaming);
+    this.responder = responder;
+    this.exceptionHandler = exceptionHandler;
 
     // The actual arguments list to invoke handler method
     this.args = new Object[args.length + 2];
-    this.args[0] = rewriteRequest(request, isStreaming);
+    this.args[0] = request;
     this.args[1] = responder;
     System.arraycopy(args, 0, this.args, 2, args.length);
   }
@@ -76,7 +81,11 @@ void invoke() throws Exception {
     } else {
       // Actually <T> would be void
       bodyConsumer = null;
-      method.invoke(handler, args);
+      try {
+        method.invoke(handler, args);
+      } catch (InvocationTargetException e) {
+        exceptionHandler.handle(e.getTargetException(), request, responder);
+      }
     }
   }
 

src/main/java/co/cask/http/HttpResourceHandler.java

@@ -60,9 +60,10 @@ public final class HttpResourceHandler implements HttpHandler {
    * @param handlers Iterable of HttpHandler.
    * @param handlerHooks Iterable of HandlerHook.
    * @param urlRewriter URL re-writer.
+   * @param exceptionHandler Exception handler
    */
   public HttpResourceHandler(Iterable<? extends HttpHandler> handlers, Iterable<? extends HandlerHook> handlerHooks,
-                             URLRewriter urlRewriter) {
+                             URLRewriter urlRewriter, ExceptionHandler exceptionHandler) {
     //Store the handlers to call init and destroy on all handlers.
     this.handlers = ImmutableList.copyOf(handlers);
     this.handlerHooks = ImmutableList.copyOf(handlerHooks);
@@ -88,7 +89,8 @@ public HttpResourceHandler(Iterable<? extends HttpHandler> handlers, Iterable<?
           Set<HttpMethod> httpMethods = getHttpMethods(method);
           Preconditions.checkArgument(httpMethods.size() >= 1,
                                       String.format("No HttpMethod found for method: %s", method.getName()));
-          patternRouter.add(absolutePath, new HttpResourceModel(httpMethods, absolutePath, method, handler));
+          patternRouter.add(absolutePath, new HttpResourceModel(httpMethods, absolutePath, method,
+                                                                handler, exceptionHandler));
         } else {
           LOG.trace("Not adding method {}({}) to path routing like. HTTP calls will not be routed to this method",
                     method.getName(), method.getParameterTypes());

src/main/java/co/cask/http/HttpResourceModel.java

@@ -55,6 +55,7 @@ public final class HttpResourceModel {
   private final Method method;
   private final HttpHandler handler;
   private final List<Map<Class<? extends Annotation>, ParameterInfo<?>>> paramsInfo;
+  private final ExceptionHandler exceptionHandler;
 
   /**
    * Construct a resource model with HttpMethod, method that handles httprequest, Object that contains the method.
@@ -64,12 +65,14 @@ public final class HttpResourceModel {
    * @param method handler that handles the http request.
    * @param handler instance {@code HttpHandler}.
    */
-  public HttpResourceModel(Set<HttpMethod> httpMethods, String path, Method method, HttpHandler handler) {
+  public HttpResourceModel(Set<HttpMethod> httpMethods, String path, Method method, HttpHandler handler,
+                           ExceptionHandler exceptionHandler) {
     this.httpMethods = httpMethods;
     this.path = path;
     this.method = method;
     this.handler = handler;
     this.paramsInfo = createParametersInfos(method);
+    this.exceptionHandler = exceptionHandler;
   }
 
   /**
@@ -131,7 +134,7 @@ public HttpMethodInfo handle(HttpRequest request, HttpResponder responder, Map<S
           idx++;
         }
 
-        return new HttpMethodInfo(method, handler, request, responder, args);
+        return new HttpMethodInfo(method, handler, request, responder, args, exceptionHandler);
       } else {
         //Found a matching resource but could not find the right HttpMethod so return 405
         throw new HandlerException(HttpResponseStatus.METHOD_NOT_ALLOWED, String.format

src/main/java/co/cask/http/NettyHttpService.java

@@ -17,6 +17,7 @@
 package co.cask.http;
 
 import com.google.common.base.Function;
+import com.google.common.base.Preconditions;
 import com.google.common.collect.ImmutableList;
 import com.google.common.collect.ImmutableMap;
 import com.google.common.collect.Maps;
@@ -42,7 +43,6 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
-import java.io.File;
 import java.net.InetSocketAddress;
 import java.util.Map;
 import java.util.concurrent.Executor;
@@ -93,6 +93,7 @@ public final class NettyHttpService extends AbstractIdleService {
    * @param urlRewriter URLRewriter to rewrite incoming URLs.
    * @param httpHandlers HttpHandlers to handle the calls.
    * @param handlerHooks Hooks to be called before/after request processing by httpHandlers.
+   *
    * @deprecated Use {@link NettyHttpService.Builder} instead.
    */
   @Deprecated
@@ -102,19 +103,9 @@ public NettyHttpService(InetSocketAddress bindAddress, int bossThreadPoolSize, i
                           RejectedExecutionHandler rejectedExecutionHandler, URLRewriter urlRewriter,
                           Iterable<? extends HttpHandler> httpHandlers,
                           Iterable<? extends HandlerHook> handlerHooks, int httpChunkLimit) {
-    this.bindAddress = bindAddress;
-    this.bossThreadPoolSize = bossThreadPoolSize;
-    this.workerThreadPoolSize = workerThreadPoolSize;
-    this.execThreadPoolSize = execThreadPoolSize;
-    this.execThreadKeepAliveSecs = execThreadKeepAliveSecs;
-    this.channelConfigs = ImmutableMap.copyOf(channelConfigs);
-    this.rejectedExecutionHandler = rejectedExecutionHandler;
-    this.channelGroup = new DefaultChannelGroup();
-    this.resourceHandler = new HttpResourceHandler(httpHandlers, handlerHooks, urlRewriter);
-    this.handlerContext = new BasicHandlerContext(this.resourceHandler);
-    this.httpChunkLimit = httpChunkLimit;
-    this.pipelineModifier = null;
-    this.sslHandlerFactory = null;
+    this(bindAddress, bossThreadPoolSize, workerThreadPoolSize, execThreadPoolSize, execThreadKeepAliveSecs,
+         channelConfigs, rejectedExecutionHandler, urlRewriter, httpHandlers, handlerHooks, httpChunkLimit,
+         null, null, new ExceptionHandler());
   }
 
   /**
@@ -131,6 +122,7 @@ public NettyHttpService(InetSocketAddress bindAddress, int bossThreadPoolSize, i
    * @param handlerHooks Hooks to be called before/after request processing by httpHandlers.
    * @param pipelineModifier Function used to modify the pipeline.
    * @param sslHandlerFactory Object used to share SSL certificate details
+   * @param exceptionHandler Handles exceptions from calling handler methods
    */
   private NettyHttpService(InetSocketAddress bindAddress, int bossThreadPoolSize, int workerThreadPoolSize,
                            int execThreadPoolSize, long execThreadKeepAliveSecs,
@@ -139,7 +131,7 @@ private NettyHttpService(InetSocketAddress bindAddress, int bossThreadPoolSize,
                            Iterable<? extends HttpHandler> httpHandlers,
                            Iterable<? extends HandlerHook> handlerHooks, int httpChunkLimit,
                            Function<ChannelPipeline, ChannelPipeline> pipelineModifier,
-                           SSLHandlerFactory sslHandlerFactory) {
+                           SSLHandlerFactory sslHandlerFactory, ExceptionHandler exceptionHandler) {
     this.bindAddress = bindAddress;
     this.bossThreadPoolSize = bossThreadPoolSize;
     this.workerThreadPoolSize = workerThreadPoolSize;
@@ -148,7 +140,7 @@ private NettyHttpService(InetSocketAddress bindAddress, int bossThreadPoolSize,
     this.channelConfigs = ImmutableMap.copyOf(channelConfigs);
     this.rejectedExecutionHandler = rejectedExecutionHandler;
     this.channelGroup = new DefaultChannelGroup();
-    this.resourceHandler = new HttpResourceHandler(httpHandlers, handlerHooks, urlRewriter);
+    this.resourceHandler = new HttpResourceHandler(httpHandlers, handlerHooks, urlRewriter, exceptionHandler);
     this.handlerContext = new BasicHandlerContext(this.resourceHandler);
     this.httpChunkLimit = httpChunkLimit;
     this.pipelineModifier = pipelineModifier;
@@ -322,6 +314,7 @@ public static class Builder {
     private int httpChunkLimit;
     private SSLHandlerFactory sslHandlerFactory;
     private Function<ChannelPipeline, ChannelPipeline> pipelineModifier;
+    private ExceptionHandler exceptionHandler;
 
     // Protected constructor to prevent instantiating Builder instance directly.
     protected Builder() {
@@ -335,6 +328,7 @@ protected Builder() {
       channelConfigs = Maps.newHashMap();
       channelConfigs.put("backlog", DEFAULT_CONNECTION_BACKLOG);
       sslHandlerFactory = null;
+      exceptionHandler = new ExceptionHandler();
     }
 
     /**
@@ -501,6 +495,12 @@ public Builder enableSSL(SSLConfig sslConfig) {
       return this;
     }
 
+    public Builder setExceptionHandler(ExceptionHandler exceptionHandler) {
+      Preconditions.checkNotNull(exceptionHandler, "exceptionHandler cannot be null");
+      this.exceptionHandler = exceptionHandler;
+      return this;
+    }
+
     /**
      * @return instance of {@code NettyHttpService}
      */
@@ -515,7 +515,7 @@ public NettyHttpService build() {
       return new NettyHttpService(bindAddress, bossThreadPoolSize, workerThreadPoolSize,
                                   execThreadPoolSize, execThreadKeepAliveSecs, channelConfigs, rejectedExecutionHandler,
                                   urlRewriter, handlers, handlerHooks, httpChunkLimit, pipelineModifier,
-                                  sslHandlerFactory);
+                                  sslHandlerFactory, exceptionHandler);
     }
   }
 }

src/test/java/co/cask/http/HttpServerTest.java

@@ -31,13 +31,16 @@
 import org.jboss.netty.channel.ChannelPipeline;
 import org.jboss.netty.handler.codec.http.HttpHeaders;
 import org.jboss.netty.handler.codec.http.HttpMethod;
+import org.jboss.netty.handler.codec.http.HttpRequest;
 import org.jboss.netty.handler.codec.http.HttpResponseStatus;
 import org.junit.AfterClass;
 import org.junit.Assert;
 import org.junit.BeforeClass;
 import org.junit.ClassRule;
 import org.junit.Test;
 import org.junit.rules.TemporaryFolder;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 import java.io.File;
 import java.io.IOException;
@@ -57,23 +60,32 @@
  */
 public class HttpServerTest {
 
+  private static final Logger LOG = LoggerFactory.getLogger(HttpServerTest.class);
+
   protected static final Type STRING_MAP_TYPE = new TypeToken<Map<String, String>>() { }.getType();
   protected static final Gson GSON = new Gson();
+  protected static final ExceptionHandler EXCEPTION_HANDLER = new ExceptionHandler() {
+    @Override
+    public void handle(Throwable t, HttpRequest request, HttpResponder responder) {
+      if (t instanceof TestHandler.CustomException) {
+        responder.sendStatus(TestHandler.CustomException.HTTP_RESPONSE_STATUS);
+      } else {
+        super.handle(t, request, responder);
+      }
+    }
+  };
 
   protected static NettyHttpService service;
   protected static URI baseURI;
 
-  @ClassRule
-  public static TemporaryFolder tmpFolder = new TemporaryFolder();
-
-  @BeforeClass
-  public static void setup() throws Exception {
+  protected static NettyHttpService.Builder createBaseNettyHttpServiceBuilder() {
     List<HttpHandler> handlers = Lists.newArrayList();
     handlers.add(new TestHandler());
 
     NettyHttpService.Builder builder = NettyHttpService.builder();
     builder.addHttpHandlers(handlers);
     builder.setHttpChunkLimit(75 * 1024);
+    builder.setExceptionHandler(EXCEPTION_HANDLER);
 
     builder.modifyChannelPipeline(new Function<ChannelPipeline, ChannelPipeline>() {
       @Override
@@ -82,8 +94,15 @@ public ChannelPipeline apply(ChannelPipeline channelPipeline) {
         return channelPipeline;
       }
     });
+    return builder;
+  }
 
-    service = builder.build();
+  @ClassRule
+  public static TemporaryFolder tmpFolder = new TemporaryFolder();
+
+  @BeforeClass
+  public static void setup() throws Exception {
+    service = createBaseNettyHttpServiceBuilder().build();
     service.startAndWait();
     Service.State state = service.state();
     Assert.assertEquals(Service.State.RUNNING, state);
@@ -305,7 +324,7 @@ public void testMultiMatchParamPut() throws Exception {
   public void testHandlerException() throws Exception {
     HttpURLConnection urlConn = request("/test/v1/uexception", HttpMethod.GET);
     Assert.assertEquals(500, urlConn.getResponseCode());
-    Assert.assertEquals("Exception Encountered while processing request : User Exception",
+    Assert.assertEquals("Exception encountered while processing request : User Exception",
                         new String(ByteStreams.toByteArray(urlConn.getErrorStream()), Charsets.UTF_8));
     urlConn.disconnect();
   }
@@ -474,6 +493,27 @@ public void testUploadReject() throws Exception {
     }
   }
 
+  @Test
+  public void testSleep() throws Exception {
+    HttpURLConnection urlConn = request("/test/v1/sleep/10", HttpMethod.GET);
+    Assert.assertEquals(200, urlConn.getResponseCode());
+    urlConn.disconnect();
+  }
+
+  @Test
+  public void testWrongMethod() throws IOException {
+    HttpURLConnection urlConn = request("/test/v1/customException", HttpMethod.GET);
+    Assert.assertEquals(HttpResponseStatus.METHOD_NOT_ALLOWED.getCode(), urlConn.getResponseCode());
+    urlConn.disconnect();
+  }
+
+  @Test
+  public void testExceptionHandler() throws IOException {
+    HttpURLConnection urlConn = request("/test/v1/customException", HttpMethod.POST);
+    Assert.assertEquals(TestHandler.CustomException.HTTP_RESPONSE_STATUS.getCode(), urlConn.getResponseCode());
+    urlConn.disconnect();
+  }
+
   protected Socket createRawSocket(URL url) throws IOException {
     return new Socket(url.getHost(), url.getPort());
   }