Merge pull request #335 from mcanlas/use-dispatcher

Use `Dispatcher` for invoking lambda handler

Author: armanbilge

build.sbt

@@ -14,6 +14,8 @@
  * limitations under the License.
  */
 
+import com.typesafe.tools.mima.core._
+
 name := "feral"
 
 ThisBuild / tlBaseVersion := "0.2"
@@ -103,6 +105,9 @@ lazy val lambda = crossProject(JSPlatform, JVMPlatform)
       "org.scalameta" %%% "munit-scalacheck" % munitVersion % Test,
       "org.typelevel" %%% "munit-cats-effect-3" % munitCEVersion % Test,
       "io.circe" %%% "circe-literal" % circeVersion % Test
+    ),
+    mimaBinaryIssueFilters ++= Seq(
+      ProblemFilters.exclude[IncompatibleResultTypeProblem]("feral.lambda.IOLambda.setupMemo")
     )
   )
   .settings(commonSettings)

core/src/main/scala/feral/IOSetup.scala

@@ -16,10 +16,13 @@
 
 package feral
 
+import scala.concurrent.Future
+
 import cats.effect.unsafe.IORuntime
 import cats.effect.kernel.Resource
 import cats.effect.IO
-import cats.effect.kernel.Deferred
+import cats.effect.std.Dispatcher
+import cats.syntax.all._
 
 private[feral] trait IOSetup {
 
@@ -28,17 +31,11 @@ private[feral] trait IOSetup {
   protected type Setup
   protected def setup: Resource[IO, Setup] = Resource.pure(null.asInstanceOf[Setup])
 
-  private[feral] final lazy val setupMemo: IO[Setup] = {
-    val deferred = Deferred.unsafe[IO, Either[Throwable, Setup]]
-    setup
-      .attempt
+  private[feral] final lazy val setupMemo: Future[(Dispatcher[IO], Setup)] =
+    (Dispatcher.parallel[IO](await = false), setup)
+      .tupled
       .allocated
-      .flatTap {
-        case (setup, _) =>
-          deferred.complete(setup)
-      }
-      .unsafeRunAndForget()(runtime)
-    deferred.get.rethrow
-  }
+      .map(_._1) // drop unused finalizer
+      .unsafeToFuture()(runtime)
 
 }

lambda/js/src/main/scala/feral/lambda/IOLambdaPlatform.scala

@@ -34,10 +34,15 @@ private[lambda] trait IOLambdaPlatform[Event, Result] {
   private lazy val handlerFn
       : js.Function2[js.Any, facade.Context, js.Promise[js.Any | Unit]] = {
     (event: js.Any, context: facade.Context) =>
-      (for {
-        lambda <- setupMemo
-        event <- IO.fromEither(decodeJs[Event](event))
-        result <- lambda(event, Context.fromJS(context))
-      } yield result.map(_.asJsAny).orUndefined).unsafeToPromise()(runtime)
+      setupMemo.toJSPromise(scala.concurrent.ExecutionContext.parasitic).`then`[js.Any] {
+        case (dispatcher, lambda) =>
+          val io =
+            for {
+              event <- IO.fromEither(decodeJs[Event](event))
+              result <- lambda(event, Context.fromJS(context))
+            } yield result.map(_.asJsAny).orUndefined
+
+          dispatcher.unsafeToPromise(io)
+      }
   }
 }

lambda/jvm/src/main/scala/feral/lambda/IOLambdaPlatform.scala

@@ -16,6 +16,9 @@
 
 package feral.lambda
 
+import scala.concurrent.Await
+import scala.concurrent.duration.Duration
+
 import cats.data.OptionT
 import cats.effect.IO
 import cats.effect.kernel.Resource
@@ -34,29 +37,33 @@ private[lambda] abstract class IOLambdaPlatform[Event, Result]
   final def handleRequest(
       input: InputStream,
       output: OutputStream,
-      context: lambdaRuntime.Context): Unit =
-    Resource
-      .eval {
-        for {
-          lambda <- setupMemo
-          event <- fs2
-            .io
-            .readInputStream(IO.pure(input), 8192, closeAfterUse = false)
-            .through(circe.fs2.byteStreamParser)
-            .through(circe.fs2.decoder[IO, Event])
-            .head
-            .compile
-            .lastOrError
-          context <- IO(Context.fromJava[IO](context))
-          _ <- OptionT(lambda(event, context)).foreachF { result =>
-            Resource.fromAutoCloseable(IO(new OutputStreamWriter(output))).use { writer =>
-              IO.blocking(Printer.noSpaces.unsafePrintToAppendable(result.asJson, writer))
+      context: lambdaRuntime.Context): Unit = {
+    val (dispatcher, lambda) =
+      Await.result(setupMemo, Duration.Inf)
+
+    dispatcher.unsafeRunSync {
+      Resource
+        .eval {
+          for {
+            event <- fs2
+              .io
+              .readInputStream(IO.pure(input), 8192, closeAfterUse = false)
+              .through(circe.fs2.byteStreamParser)
+              .through(circe.fs2.decoder[IO, Event])
+              .head
+              .compile
+              .lastOrError
+            context <- IO(Context.fromJava[IO](context))
+            _ <- OptionT(lambda(event, context)).foreachF { result =>
+              Resource.fromAutoCloseable(IO(new OutputStreamWriter(output))).use { writer =>
+                IO.blocking(Printer.noSpaces.unsafePrintToAppendable(result.asJson, writer))
+              }
             }
-          }
-        } yield ()
-      }
-      .onFinalize(IO.blocking(input.close()) &> IO.blocking(output.close()))
-      .use_
-      .unsafeRunSync()(runtime)
+          } yield ()
+        }
+        .onFinalize(IO.blocking(input.close()) &> IO.blocking(output.close()))
+        .use_
+    }
+  }
 
 }

lambda/shared/src/test/scala/feral/lambda/IOLambdaSuite.scala

@@ -34,7 +34,9 @@ class IOLambdaSuite extends CatsEffectSuite {
             .as(_.event.map(Some(_)) <* invokeCounter.getAndUpdate(_ + 1))
         }
       }
-      handler <- lambda.setupMemo
+
+      handler <- IO.fromFuture(IO(lambda.setupMemo)).map(_._2)
+
       _ <- ('0' to 'z')
         .map(_.toString)
         .toList