Make Lambda generic in effect type

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

@@ -16,30 +16,45 @@
 
 package feral
 
-import cats.effect.unsafe.IORuntime
-import cats.effect.kernel.Resource
 import cats.effect.IO
+import cats.effect.kernel.Async
 import cats.effect.kernel.Deferred
+import cats.effect.kernel.Resource
+import cats.effect.std.Dispatcher
+import cats.effect.unsafe.IORuntime
 import cats.syntax.all._
 
-private[feral] trait IOSetup {
+import scala.concurrent.Future
 
-  protected def runtime: IORuntime = IORuntime.global
+trait Feral[F[_]] {
+  implicit protected def F: Async[F]
+
+  protected def dispatcher: Dispatcher[F]
 
   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
-      .allocated
-      .flatTap {
+  protected def setup: Resource[F, Setup] = Resource.pure(null.asInstanceOf[Setup])
+
+  final lazy val setupMemo: F[Setup] = {
+    val deferred = Deferred.unsafe[F, Either[Throwable, Setup]]
+    dispatcher.unsafeRunAndForget {
+      setup.attempt.allocated.flatTap {
         case (setup, _) =>
           deferred.complete(setup)
       }
-      .unsafeRunAndForget()(runtime)
+    }
     deferred.get.rethrow
   }
 
 }
+
+trait IOFeral extends Feral[IO] {
+
+  protected def runtime: IORuntime = IORuntime.global
+
+  implicit protected final def F = IO.asyncForIO
+
+  protected final def dispatcher = new Dispatcher[IO] {
+    override def unsafeToFutureCancelable[A](fa: IO[A]): (Future[A], () => Future[Unit]) =
+      fa.unsafeToFutureCancelable()(runtime)
+  }
+}

lambda-api-gateway-proxy-http4s/src/main/scala/feral/lambda/ApiGatewayProxyHttp4sLambda.scala

@@ -16,9 +16,9 @@
 
 package feral.lambda
 
-import cats.effect.IO
 import cats.effect.Resource
 import cats.effect.SyncIO
+import cats.syntax.all._
 import feral.lambda.events.ApiGatewayProxyEventV2
 import feral.lambda.events.ApiGatewayProxyStructuredResultV2
 import fs2.Stream
@@ -33,37 +33,37 @@ import org.http4s.Uri
 import org.typelevel.vault.Key
 import org.typelevel.vault.Vault
 
-abstract class ApiGatewayProxyHttp4sLambda
-    extends IOLambda[ApiGatewayProxyEventV2, ApiGatewayProxyStructuredResultV2] {
+abstract class ApiGatewayProxyHttp4sLambda[F[_]]
+    extends Lambda[F, ApiGatewayProxyEventV2, ApiGatewayProxyStructuredResultV2] {
 
-  val ContextKey = Key.newKey[SyncIO, Context].unsafeRunSync()
+  val ContextKey = Key.newKey[SyncIO, Context[F]].unsafeRunSync()
   val EventKey = Key.newKey[SyncIO, ApiGatewayProxyEventV2].unsafeRunSync()
 
-  def routes: Resource[IO, HttpRoutes[IO]]
+  def routes: Resource[F, HttpRoutes[F]]
 
-  protected type Setup = HttpRoutes[IO]
-  protected override final val setup: Resource[IO, HttpRoutes[IO]] = routes
+  protected type Setup = HttpRoutes[F]
+  protected override final val setup: Resource[F, HttpRoutes[F]] = routes
 
   override final def apply(
       event: ApiGatewayProxyEventV2,
-      context: Context,
-      routes: HttpRoutes[IO]): IO[Some[ApiGatewayProxyStructuredResultV2]] =
+      context: Context[F],
+      routes: HttpRoutes[F]): F[Option[ApiGatewayProxyStructuredResultV2]] =
     for {
-      method <- IO.fromEither(Method.fromString(event.requestContext.http.method))
-      uri <- IO.fromEither(Uri.fromString(event.rawPath))
+      method <- Method.fromString(event.requestContext.http.method).liftTo
+      uri <- Uri.fromString(event.rawPath).liftTo
       headers = Headers(event.headers.toList)
       requestBody =
         if (event.isBase64Encoded)
-          Stream.fromOption[IO](event.body).through(fs2.text.base64.decode)
+          Stream.fromOption[F](event.body).through(fs2.text.base64.decode)
         else
-          Stream.fromOption[IO](event.body).through(fs2.text.utf8.encode)
+          Stream.fromOption[F](event.body).through(fs2.text.utf8.encode)
       request = Request(
         method,
         uri,
         headers = headers,
         body = requestBody,
         attributes = Vault.empty.insert(ContextKey, context).insert(EventKey, event))
-      response <- routes(request).getOrElse(Response.notFound[IO])
+      response <- routes(request).getOrElse(Response.notFound[F])
       isBase64Encoded = !response.charset.contains(Charset.`UTF-8`)
       responseBody <- (if (isBase64Encoded)
                          response.body.through(fs2.text.base64.encode)

lambda-cloudformation-custom-resource/src/main/scala/feral/lambda/cloudformation/CloudFormationCustomResource.scala

@@ -19,47 +19,56 @@ package feral.lambda.cloudformation
 import cats.effect._
 import cats.effect.kernel.Resource
 import cats.syntax.all._
+import feral.IOFeral
 import feral.lambda.cloudformation.CloudFormationCustomResourceHandler.stackTraceLines
 import feral.lambda.cloudformation.CloudFormationRequestType._
-import feral.lambda.{Context, IOLambda}
+import feral.lambda.{Context, Lambda}
 import io.circe._
 import io.circe.syntax._
 import org.http4s.Method.POST
-import org.http4s.client.Client
 import org.http4s.circe._
-import org.http4s.client.dsl.io._
+import org.http4s.client.Client
+import org.http4s.client.dsl.Http4sClientDsl
 import org.http4s.ember.client.EmberClientBuilder
 
 import java.io.{PrintWriter, StringWriter}
 
 trait CloudFormationCustomResource[F[_], Input, Output] {
   def createResource(
       event: CloudFormationCustomResourceRequest[Input],
-      context: Context): F[HandlerResponse[Output]]
+      context: Context[F]): F[HandlerResponse[Output]]
   def updateResource(
       event: CloudFormationCustomResourceRequest[Input],
-      context: Context): F[HandlerResponse[Output]]
+      context: Context[F]): F[HandlerResponse[Output]]
   def deleteResource(
       event: CloudFormationCustomResourceRequest[Input],
-      context: Context): F[HandlerResponse[Output]]
+      context: Context[F]): F[HandlerResponse[Output]]
 }
 
-abstract class CloudFormationCustomResourceHandler[Input: Decoder, Output: Encoder]
-    extends IOLambda[CloudFormationCustomResourceRequest[Input], Unit] {
-  type Setup = (Client[IO], CloudFormationCustomResource[IO, Input, Output])
+abstract class IOCloudFormationCustomResourceHandler[Input: Decoder, Output: Encoder]
+    extends CloudFormationCustomResourceHandler[IO, Input, Output]
+    with IOFeral
+
+abstract class CloudFormationCustomResourceHandler[F[_], Input: Decoder, Output: Encoder]
+    extends Lambda[F, CloudFormationCustomResourceRequest[Input], Unit] {
+
+  private val http4sClientDsl = new Http4sClientDsl[F] {}
+  import http4sClientDsl._
+
+  type Setup = (Client[F], CloudFormationCustomResource[F, Input, Output])
 
-  override final def setup: Resource[IO, Setup] =
+  override final def setup: Resource[F, Setup] =
     client.mproduct(handler)
 
-  protected def client: Resource[IO, Client[IO]] =
-    EmberClientBuilder.default[IO].build
+  protected def client: Resource[F, Client[F]] =
+    EmberClientBuilder.default[F].build
 
-  def handler(client: Client[IO]): Resource[IO, CloudFormationCustomResource[IO, Input, Output]]
+  def handler(client: Client[F]): Resource[F, CloudFormationCustomResource[F, Input, Output]]
 
   override def apply(
       event: CloudFormationCustomResourceRequest[Input],
-      context: Context,
-      setup: Setup): IO[Option[Unit]] =
+      context: Context[F],
+      setup: Setup): F[Option[Unit]] =
     (event.RequestType match {
       case CreateRequest => setup._2.createResource(event, context)
       case UpdateRequest => setup._2.updateResource(event, context)
@@ -70,9 +79,9 @@ abstract class CloudFormationCustomResourceHandler[Input: Decoder, Output: Encod
       .flatMap { resp => setup._1.successful(POST(resp.asJson, event.ResponseURL)) }
       .as(None)
 
-  private def illegalRequestType[A](other: String): IO[A] =
+  private def illegalRequestType[A](other: String): F[A] =
     (new IllegalArgumentException(
-      s"unexpected CloudFormation request type `$other``"): Throwable).raiseError[IO, A]
+      s"unexpected CloudFormation request type `$other``"): Throwable).raiseError[F, A]
 
   private def exceptionResponse(req: CloudFormationCustomResourceRequest[Input])(
       ex: Throwable): CloudFormationCustomResourceResponse =

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

@@ -16,13 +16,13 @@
 
 package feral.lambda
 
-import cats.effect.IO
+import cats.effect.kernel.Sync
 
 import scala.concurrent.duration._
 
 private[lambda] trait ContextCompanionPlatform {
 
-  private[lambda] def fromJS(context: facade.Context): Context =
+  private[lambda] def fromJS[F[_]](context: facade.Context)(implicit F: Sync[F]): Context[F] =
     Context(
       context.functionName,
       context.functionVersion,
@@ -55,6 +55,6 @@ private[lambda] trait ContextCompanionPlatform {
             )
           )
         },
-      IO(context.getRemainingTimeInMillis().millis)
+      F.delay(context.getRemainingTimeInMillis().millis)
     )
 }

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

@@ -16,21 +16,23 @@
 
 package feral.lambda
 
-import cats.effect.IO
+import cats.syntax.all._
 import io.circe.scalajs._
 
 import scala.scalajs.js
 import scala.scalajs.js.JSConverters._
 import scala.scalajs.js.|
 
-private[lambda] trait IOLambdaPlatform[Event, Result] {
-  this: IOLambda[Event, Result] =>
+private[lambda] trait LambdaPlatform[F[_], Event, Result] {
+  this: Lambda[F, Event, Result] =>
 
   // @JSExportTopLevel("handler") // TODO
   final def handler(event: js.Any, context: facade.Context): js.Promise[js.Any | Unit] =
-    (for {
-      setup <- setupMemo
-      event <- IO.fromEither(decodeJs[Event](event))
-      result <- apply(event, Context.fromJS(context), setup)
-    } yield result.map(_.asJsAny).orUndefined).unsafeToPromise()(runtime)
+    dispatcher.unsafeToPromise {
+      for {
+        setup <- setupMemo
+        event <- decodeJs[Event](event).liftTo
+        result <- apply(event, Context.fromJS(context), setup)
+      } yield result.map(_.asJsAny).orUndefined
+    }
 }

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

@@ -16,14 +16,15 @@
 
 package feral.lambda
 
-import cats.effect.IO
+import cats.effect.kernel.Sync
 import com.amazonaws.services.lambda.runtime
 
 import scala.concurrent.duration._
 
 private[lambda] trait ContextCompanionPlatform {
 
-  private[lambda] def fromJava(context: runtime.Context): Context =
+  private[lambda] def fromJava[F[_]](context: runtime.Context)(
+      implicit F: Sync[F]): Context[F] =
     Context(
       context.getFunctionName(),
       context.getFunctionVersion(),
@@ -53,7 +54,7 @@ private[lambda] trait ContextCompanionPlatform {
           )
         )
       },
-      IO(context.getRemainingTimeInMillis().millis)
+      F.delay(context.getRemainingTimeInMillis().millis)
     )
 
 }

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

@@ -17,44 +17,44 @@
 package feral.lambda
 
 import cats.data.OptionT
-import cats.effect.IO
 import cats.effect.kernel.Resource
+import cats.syntax.all._
 import com.amazonaws.services.lambda.{runtime => lambdaRuntime}
 import io.circe
 
 import java.io.InputStream
 import java.io.OutputStream
 import java.nio.charset.StandardCharsets
 
-private[lambda] abstract class IOLambdaPlatform[Event, Result]
-    extends lambdaRuntime.RequestStreamHandler { this: IOLambda[Event, Result] =>
+private[lambda] abstract class LambdaPlatform[F[_], Event, Result]
+    extends lambdaRuntime.RequestStreamHandler { this: Lambda[F, Event, Result] =>
 
   final def handleRequest(
       input: InputStream,
       output: OutputStream,
-      context: lambdaRuntime.Context): Unit =
+      context: lambdaRuntime.Context): Unit = dispatcher.unsafeRunSync {
     Resource
       .eval {
         for {
           setup <- setupMemo
           event <- fs2
             .io
-            .readInputStream(IO.pure(input), 8192, closeAfterUse = false)
+            .readInputStream(input.pure, 8192, closeAfterUse = false)
             .through(circe.fs2.byteStreamParser)
-            .through(circe.fs2.decoder[IO, Event])
+            .through(circe.fs2.decoder[F, Event])
             .head
             .compile
             .lastOrError
-          context <- IO(Context.fromJava(context))
-          _ <- OptionT(apply(event, context, setup)).foldF(IO.unit) { result =>
+          context <- F.delay(Context.fromJava(context))
+          _ <- OptionT(apply(event, context, setup)).foldF(F.unit) { result =>
             // TODO can circe write directly to output?
-            IO(output.write(encoder(result).noSpaces.getBytes(StandardCharsets.UTF_8)))
+            F.delay(output.write(encoder(result).noSpaces.getBytes(StandardCharsets.UTF_8)))
           }
         } yield ()
       }
-      .onFinalize(IO(input.close()))
-      .onFinalize(IO(output.close()))
+      .onFinalize(F.delay(input.close()))
+      .onFinalize(F.delay(output.close()))
       .use_
-      .unsafeRunSync()(runtime)
+  }
 
 }

lambda/shared/src/main/scala/feral/lambda/Context.scala

@@ -16,11 +16,9 @@
 
 package feral.lambda
 
-import cats.effect.IO
-
 import scala.concurrent.duration.FiniteDuration
 
-final case class Context(
+final case class Context[F[_]](
     functionName: String,
     functionVersion: String,
     invokedFunctionArn: String,
@@ -30,7 +28,7 @@ final case class Context(
     logStreamName: String,
     identity: Option[CognitoIdentity],
     clientContext: Option[ClientContext],
-    remainingTime: IO[FiniteDuration]
+    remainingTime: F[FiniteDuration]
 )
 
 object Context extends ContextCompanionPlatform

lambda/shared/src/main/scala/feral/lambda/IOLambda.scala → lambda/shared/src/main/scala/feral/lambda/Lambda.scala

@@ -21,11 +21,13 @@ import cats.effect.IO
 import io.circe.Decoder
 import io.circe.Encoder
 
-abstract class IOLambda[Event, Result](
+abstract class Lambda[F[_], Event, Result](
     implicit private[lambda] val decoder: Decoder[Event],
     private[lambda] val encoder: Encoder[Result]
-) extends IOLambdaPlatform[Event, Result]
-    with IOSetup {
+) extends LambdaPlatform[F, Event, Result]
+    with Feral[F] {
 
-  def apply(event: Event, context: Context, setup: Setup): IO[Option[Result]]
+  def apply(event: Event, context: Context[F], setup: Setup): F[Option[Result]]
 }
+
+trait IOLambda[Event, Result] extends Lambda[IO, Event, Result] with IOFeral