airframe-http: Add HttpFilter support to Router (#540)

* Reorganize Router structure
* Adding before/after filter
* Pass the surface of a request filter to use DI
* Use immutable Route implementation
* Use tree structure for Router
* Add workaround for slf4j flush delay
* Add FinagleFilter, FinagleContext

airframe-http-finagle/src/main/scala/wvlet/airframe/http/finagle/FinagleResponseHandler.scala

@@ -0,0 +1,81 @@
+/*
+ * 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 wvlet.airframe.http.finagle
+
+import com.twitter.finagle.http.{Request, Response, Status}
+import com.twitter.io.Buf.ByteArray
+import wvlet.airframe.codec.{JSONCodec, MessageCodec, MessageCodecFactory}
+import wvlet.airframe.http.{ResponseHandler, SimpleHttpResponse}
+import wvlet.airframe.surface.Surface
+
+/**
+  * Converting controller results into finagle http responses.
+  */
+trait FinagleResponseHandler extends ResponseHandler[Request, Response] {
+
+  // Use Map codecs to create natural JSON responses
+  private[this] val mapCodecFactory =
+    MessageCodecFactory.defaultFactory.withObjectMapCodec
+
+  // TODO: Extract this logic into airframe-http
+  def toHttpResponse[A](request: Request, responseSurface: Surface, a: A): Response = {
+    a match {
+      case r: Response =>
+        // Return the response as is
+        r
+      case r: SimpleHttpResponse =>
+        val resp = Response(request)
+        resp.statusCode = r.statusCode
+        resp.contentString = r.contentString
+        r.contentType.map { c =>
+          resp.contentType = c
+        }
+        resp
+      case s: String =>
+        val r = Response()
+        r.contentString = s
+        r
+      case _ =>
+        // TODO Return large responses with streams
+
+        // Convert the response object into JSON
+        val rs = mapCodecFactory.of(responseSurface)
+        val msgpack: Array[Byte] = rs match {
+          case m: MessageCodec[_] =>
+            m.asInstanceOf[MessageCodec[A]].toMsgPack(a)
+          case _ =>
+            throw new IllegalArgumentException(s"Unknown codec: ${rs}")
+        }
+
+        // Return application/msgpack content type
+        if (request.accept.contains("application/x-msgpack")) {
+          val res = Response(Status.Ok)
+          res.contentType = "application/x-msgpack"
+          res.content = ByteArray.Owned(msgpack)
+          res
+        } else {
+          val json = JSONCodec.unpackMsgPack(msgpack)
+          json match {
+            case Some(j) =>
+              val res = Response(Status.Ok)
+              res.setContentTypeJson()
+              res.setContentString(json.get)
+              res
+            case None =>
+              Response(Status.InternalServerError)
+          }
+        }
+    }
+  }
+}

airframe-http-finagle/src/main/scala/wvlet/airframe/http/finagle/FinagleRouter.scala

@@ -13,111 +13,115 @@
  */
 package wvlet.airframe.http.finagle
 
-import com.twitter.finagle.http.{Request, Response, Status}
+import com.twitter.finagle.http.{Request, Response}
 import com.twitter.finagle.{Service, SimpleFilter}
-import com.twitter.io.Buf.ByteArray
 import com.twitter.util.Future
-import wvlet.airframe.codec.{JSONCodec, MessageCodec, MessageCodecFactory}
-import wvlet.airframe.http.{ControllerProvider, ResponseHandler}
-import wvlet.airframe.surface.Surface
+import wvlet.airframe.http._
+import wvlet.airframe.http.finagle.FinagleRouter.RouteFilter
 import wvlet.log.LogSupport
 
 /**
-  * A filter for dispatching http requests to the predefined routes with Finagle
+  * A router for dispatching http requests to the predefined routes.
   */
 class FinagleRouter(config: FinagleServerConfig,
                     controllerProvider: ControllerProvider,
                     responseHandler: ResponseHandler[Request, Response])
     extends SimpleFilter[Request, Response]
     with LogSupport {
 
+  // A table for Route -> matching HttpFilter
+  private val filterTable: Map[Route, RouteFilter] =
+    FinagleRouter.buildRouteFilters(config.router, FinagleRouter.identityFilter, controllerProvider)
+
   override def apply(request: Request, service: Service[Request, Response]): Future[Response] = {
     // Find a route matching to the request
     config.router.findRoute(request) match {
       case Some(routeMatch) =>
-        val route = routeMatch.route
-        // Find a corresponding controller
-        controllerProvider.findController(route.controllerSurface) match {
-          case Some(controller) =>
-            // Call the method in this controller
-            val args   = route.buildControllerMethodArgs(controller, request, routeMatch.params)
-            val result = route.call(controller, args)
-
-            route.returnTypeSurface.rawType match {
-              // When a return type is Future[X]
-              case f if classOf[Future[_]].isAssignableFrom(f) =>
-                // Check the type of X
-                val futureValueSurface = route.returnTypeSurface.typeArgs(0)
-                futureValueSurface.rawType match {
-                  // If X is Response type, return as is
-                  case vc if classOf[Response].isAssignableFrom(vc) =>
-                    result.asInstanceOf[Future[Response]]
-                  case other =>
-                    // If X is other type, convert X into an HttpResponse
-                    result.asInstanceOf[Future[_]].map { r =>
-                      responseHandler.toHttpResponse(request, futureValueSurface, r)
-                    }
-                }
-              case _ =>
-                // If the route returns non future value, convert it into Future response
-                Future.value(responseHandler.toHttpResponse(request, route.returnTypeSurface, result))
-            }
-          case None =>
-            Future.exception(new IllegalStateException(s"Controller ${route.controllerSurface} is not found"))
-        }
+        val routeFilter    = filterTable(routeMatch.route)
+        val context        = new FinagleRouter.FinagleHttpContext(routeMatch, routeFilter.controller, responseHandler)
+        val currentService = routeFilter.filter.andThen(context)
+        currentService(request)
       case None =>
         // No route is found
         service(request)
     }
   }
 }
 
-/**
-  * Converting controller results into finagle http responses.
-  */
-trait FinagleResponseHandler extends ResponseHandler[Request, Response] {
+object FinagleRouter {
+
+  case object Identity extends HttpFilter.Identity[Request, Response, Future]
+  def identityFilter = Identity
+
+  case class RouteFilter(filter: FinagleFilter, controller: Any)
 
-  // Use Map codecs to create natural JSON responses
-  private[this] val mapCodecFactory =
-    MessageCodecFactory.defaultFactory.withObjectMapCodec
+  /**
+    * Traverse the Router tree and build HttpFilter for each local Route
+    */
+  def buildRouteFilters(router: Router,
+                        parentFilter: FinagleFilter,
+                        controllerProvider: ControllerProvider): Map[Route, RouteFilter] = {
 
-  def toHttpResponse[A](request: Request, responseSurface: Surface, a: A): Response = {
-    a match {
-      case r: Response =>
-        // Return the response as is
-        r
-      case s: String =>
-        val r = Response()
-        r.contentString = s
-        r
-      case _ =>
-        // Convert the response object into JSON
-        val rs = mapCodecFactory.of(responseSurface)
-        val msgpack: Array[Byte] = rs match {
-          case m: MessageCodec[_] =>
-            m.asInstanceOf[MessageCodec[A]].toMsgPack(a)
-          case _ =>
-            throw new IllegalArgumentException(s"Unknown codec: ${rs}")
+    // TODO Extract this logic into airframe-http
+    val localFilterOpt: Option[FinagleFilter] =
+      router.filterSurface
+        .map(fs => controllerProvider.findController(fs))
+        .filter(_.isDefined)
+        // TODO convert generic http filter to FinagleFilter
+        .map(_.get.asInstanceOf[FinagleFilter])
+
+    val currentFilter: FinagleFilter =
+      localFilterOpt
+        .map { l =>
+          parentFilter.andThen(l)
         }
+        .getOrElse(parentFilter)
 
-        // TODO return application/msgpack content type
-        if (request.accept.contains("application/x-msgpack")) {
-          val res = Response(Status.Ok)
-          res.contentType = "application/x-msgpack"
-          res.content = ByteArray.Owned(msgpack)
-          res
-        } else {
-          val json = JSONCodec.unpackMsgPack(msgpack)
-          json match {
-            case Some(j) =>
-              val res = Response(Status.Ok)
-              res.setContentTypeJson()
-              res.setContentString(json.get)
-              res
-            case None =>
-              Response(Status.InternalServerError)
+    val m = Map.newBuilder[Route, RouteFilter]
+    for (route <- router.localRoutes) {
+      val controller = controllerProvider.findController(route.controllerSurface)
+      if (controller.isEmpty) {
+        throw new IllegalStateException(s"Missing controller. Add ${route.controllerSurface} to the design")
+      }
+      m += (route -> RouteFilter(currentFilter, controller.get))
+    }
+    for (c <- router.children) {
+      m ++= buildRouteFilters(c, currentFilter, controllerProvider)
+    }
+    m.result()
+  }
+
+  /**
+    *  Call a controller method by mapping the request parameters to the method arguments.
+    *  This will be the last context after applying preceding filters
+    */
+  class FinagleHttpContext(routeMatch: RouteMatch, controller: Any, responseHandler: ResponseHandler[Request, Response])
+      extends HttpContext[Request, Response, Future] {
+    override def apply(request: Request): Future[Response] = {
+      val route = routeMatch.route
+      // Call the method in this controller
+      val args   = route.buildControllerMethodArgs(controller, request, routeMatch.params)
+      val result = route.call(controller, args)
+
+      route.returnTypeSurface.rawType match {
+        // When a return type is Future[X]
+        case f if classOf[Future[_]].isAssignableFrom(f) =>
+          // Check the type of X
+          val futureValueSurface = route.returnTypeSurface.typeArgs(0)
+          futureValueSurface.rawType match {
+            // If X is Response type, return as is
+            case vc if classOf[Response].isAssignableFrom(vc) =>
+              result.asInstanceOf[Future[Response]]
+            case other =>
+              // If X is other type, convert X into an HttpResponse
+              result.asInstanceOf[Future[_]].map { r =>
+                responseHandler.toHttpResponse(request, futureValueSurface, r)
+              }
           }
-        }
+        case _ =>
+          // If the route returns non future value, convert it into Future response
+          Future.value(responseHandler.toHttpResponse(request, route.returnTypeSurface, result))
+      }
     }
   }
 }

airframe-http-finagle/src/main/scala/wvlet/airframe/http/finagle/package.scala

@@ -16,6 +16,7 @@ package wvlet.airframe.http
 import com.twitter.finagle.http
 import com.twitter.finagle.http.{Request, Response}
 import com.twitter.io.Buf.ByteArray
+import com.twitter.util.Future
 import wvlet.airframe.Design
 import wvlet.airframe.http.finagle.FinagleServer.FinagleService
 import wvlet.log.io.IOUtil
@@ -25,6 +26,9 @@ import wvlet.log.io.IOUtil
   */
 package object finagle {
 
+  type FinagleFilter  = HttpFilter[Request, Response, Future]
+  type FinagleContext = HttpContext[Request, Response, Future]
+
   private def finagleBaseDesign: Design =
     httpDefaultDesign
       .bind[ResponseHandler[http.Request, http.Response]].to[FinagleResponseHandler]
@@ -52,10 +56,11 @@ package object finagle {
   }
 
   implicit object FinagleHttpRequestAdapter extends HttpRequestAdapter[http.Request] {
-    override def methodOf(request: Request): HttpMethod         = toHttpMethod(request.method)
-    override def pathOf(request: Request): String               = request.path
-    override def queryOf(request: Request): Map[String, String] = request.params
-    override def contentStringOf(request: Request): String      = request.contentString
+    override def methodOf(request: Request): HttpMethod          = toHttpMethod(request.method)
+    override def pathOf(request: Request): String                = request.path
+    override def headerOf(request: Request): Map[String, String] = request.headerMap.toMap
+    override def queryOf(request: Request): Map[String, String]  = request.params
+    override def contentStringOf(request: Request): String       = request.contentString
     override def contentBytesOf(request: Request): Array[Byte] = {
       val content = request.content
       val size    = content.length