netty (fix): Fix async REST response handling (Scala Future, error Rx)

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

@@ -37,7 +37,8 @@ class FinagleRouter(session: Session, private[finagle] val config: FinagleServer
       config.controllerProvider,
       FinagleBackend,
       config.responseHandler,
-      MessageCodecFactory.defaultFactory.orElse(MessageCodecFactory.newFactory(config.customCodec))
+      MessageCodecFactory.defaultFactory.orElse(MessageCodecFactory.newFactory(config.customCodec)),
+      config.executionContext
     )
 
   override def apply(request: Request, service: Service[Request, Response]): Future[Response] = {

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

@@ -22,7 +22,7 @@ import com.twitter.util.{Await, Future}
 import javax.annotation.PostConstruct
 import wvlet.airframe.*
 import wvlet.airframe.codec.MessageCodec
-import wvlet.airframe.control.MultipleExceptions
+import wvlet.airframe.control.{MultipleExceptions, ThreadUtil}
 import wvlet.airframe.http.finagle.FinagleServer.FinagleService
 import wvlet.airframe.http.finagle.filter.HttpAccessLogFilter
 import wvlet.airframe.http.{
@@ -40,9 +40,10 @@ import wvlet.airframe.surface.Surface
 import wvlet.log.LogSupport
 import wvlet.log.io.IOUtil
 
+import java.util.concurrent.Executors
 import scala.annotation.tailrec
 import scala.collection.parallel.immutable.ParVector
-import scala.concurrent.ExecutionException
+import scala.concurrent.{ExecutionContext, ExecutionException}
 import scala.util.Try
 import scala.util.control.NonFatal
 
@@ -62,7 +63,15 @@ case class FinagleServerConfig(
     // A top-level filter applied before routing requests
     beforeRoutingFilter: Filter[Request, Response, Request, Response] = Filter.identity,
     // Service called when no matching route is found
-    fallbackService: Service[Request, Response] = FinagleServer.notFound
+    fallbackService: Service[Request, Response] = FinagleServer.notFound,
+    // Thread manager for handling Future[_] responses
+    executionContext: ExecutionContext = {
+      // Using the global thread pool causes an issue in sbt's layered class loader #918
+      // So need to use the local daemon thread pool
+      ExecutionContext.fromExecutorService(
+        Executors.newCachedThreadPool(ThreadUtil.newDaemonThreadFactory("airframe-finagle"))
+      )
+    }
 ) {
   // Lazily acquire an unused port to avoid conflicts between multiple servers
   lazy val port = serverPort.getOrElse(IOUtil.unusedPort)

airframe-http-netty/src/main/scala/wvlet/airframe/http/netty/NettyBackend.scala

@@ -19,7 +19,8 @@ import wvlet.airframe.rx.Rx
 import wvlet.log.LogSupport
 
 import scala.collection.mutable
-import scala.concurrent.{ExecutionContext, Future, Promise}
+import scala.concurrent.{Await, ExecutionContext, Future, Promise}
+import scala.util.{Failure, Success}
 
 object NettyBackend extends HttpBackend[Request, Response, Rx] with LogSupport { self =>
   private val rxBackend = new RxNettyBackend
@@ -37,17 +38,18 @@ object NettyBackend extends HttpBackend[Request, Response, Rx] with LogSupport {
     Rx.single(a)
   }
 
-  override def toFuture[A](a: Future[A], e: ExecutionContext): Rx[A] = {
-    Rx.future(a)(e)
+  override def toFuture[A](a: Future[A], ex: ExecutionContext): Rx[A] = {
+    val v = Await.result(a, scala.concurrent.duration.Duration.Inf)
+    Rx.single(v)
   }
 
   override def toScalaFuture[A](a: Rx[A]): Future[A] = {
     val promise: Promise[A] = Promise()
-    a.toRx
-      .map { x =>
-        promise.success(x)
-      }
-      .recover { case e: Throwable => promise.failure(e) }
+    val rx = a.transform {
+      case Success(x)  => promise.success(x)
+      case Failure(ex) => promise.failure(ex)
+    }
+    rx.run { effect => }
     promise.future
   }
 

airframe-http-netty/src/main/scala/wvlet/airframe/http/netty/NettyServer.scala

@@ -35,10 +35,11 @@ import wvlet.airframe.{Design, Session}
 import wvlet.log.LogSupport
 import wvlet.log.io.IOUtil
 
-import java.util.concurrent.TimeUnit
+import java.util.concurrent.{Executors, TimeUnit}
 import java.util.concurrent.atomic.AtomicBoolean
 import javax.annotation.PostConstruct
 import scala.collection.immutable.ListMap
+import scala.concurrent.ExecutionContext
 import scala.util.{Failure, Success, Try}
 
 case class NettyServerConfig(
@@ -52,7 +53,15 @@ case class NettyServerConfig(
       new LogRotationHttpLogger(config)
     },
     loggingFilter: HttpLogger => RxHttpFilter = { new RPCLoggingFilter(_) },
-    customCodec: PartialFunction[Surface, MessageCodec[_]] = PartialFunction.empty
+    customCodec: PartialFunction[Surface, MessageCodec[_]] = PartialFunction.empty,
+    // Thread manager for handling Future[_] responses
+    executionContext: ExecutionContext = {
+      // Using the global thread pool causes an issue in sbt's layered class loader #918
+      // So need to use the local daemon thread pool
+      ExecutionContext.fromExecutorService(
+        Executors.newCachedThreadPool(ThreadUtil.newDaemonThreadFactory("airframe-netty"))
+      )
+    }
 ) {
   lazy val port = serverPort.getOrElse(IOUtil.unusedPort)
 
@@ -149,7 +158,6 @@ class NettyServer(config: NettyServerConfig, session: Session) extends HttpServe
       new NioEventLoopGroup(numWorkers, tf)
     }
   }
-
   private var channelFuture: Option[Channel] = None
 
   override def localAddress: String = s"localhost:${config.port}"
@@ -238,7 +246,8 @@ class NettyServer(config: NettyServerConfig, session: Session) extends HttpServe
               NettyBackend,
               new NettyResponseHandler,
               // Set a custom codec and use JSON map output
-              MessageCodecFactory.defaultFactoryForJSON.withCodecs(config.customCodec)
+              MessageCodecFactory.defaultFactoryForJSON.withCodecs(config.customCodec),
+              config.executionContext
             )
           )
       }