diff --git a/core/src/main/scala/feral/IOSetup.scala b/core/src/main/scala/feral/Feral.scala similarity index 55% rename from core/src/main/scala/feral/IOSetup.scala rename to core/src/main/scala/feral/Feral.scala index eda4dbe8..9159bb87 100644 --- a/core/src/main/scala/feral/IOSetup.scala +++ b/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) + } +} diff --git a/lambda-api-gateway-proxy-http4s/src/main/scala/feral/lambda/ApiGatewayProxyHttp4sLambda.scala b/lambda-api-gateway-proxy-http4s/src/main/scala/feral/lambda/ApiGatewayProxyHttp4sLambda.scala index 06d1bce9..6c8d6855 100644 --- a/lambda-api-gateway-proxy-http4s/src/main/scala/feral/lambda/ApiGatewayProxyHttp4sLambda.scala +++ b/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) diff --git a/lambda-cloudformation-custom-resource/src/main/scala/feral/lambda/cloudformation/CloudFormationCustomResource.scala b/lambda-cloudformation-custom-resource/src/main/scala/feral/lambda/cloudformation/CloudFormationCustomResource.scala index 04cda301..f8898b27 100644 --- a/lambda-cloudformation-custom-resource/src/main/scala/feral/lambda/cloudformation/CloudFormationCustomResource.scala +++ b/lambda-cloudformation-custom-resource/src/main/scala/feral/lambda/cloudformation/CloudFormationCustomResource.scala @@ -19,15 +19,16 @@ 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} @@ -35,31 +36,39 @@ 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 = diff --git a/lambda/js/src/main/scala/feral/lambda/ContextPlatform.scala b/lambda/js/src/main/scala/feral/lambda/ContextPlatform.scala index 8d167b01..2fd63c68 100644 --- a/lambda/js/src/main/scala/feral/lambda/ContextPlatform.scala +++ b/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) ) } diff --git a/lambda/js/src/main/scala/feral/lambda/IOLambdaPlatform.scala b/lambda/js/src/main/scala/feral/lambda/LambdaPlatform.scala similarity index 70% rename from lambda/js/src/main/scala/feral/lambda/IOLambdaPlatform.scala rename to lambda/js/src/main/scala/feral/lambda/LambdaPlatform.scala index d71bbeb9..f67d6c52 100644 --- a/lambda/js/src/main/scala/feral/lambda/IOLambdaPlatform.scala +++ b/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 + } } diff --git a/lambda/jvm/src/main/scala/feral/lambda/ContextPlatform.scala b/lambda/jvm/src/main/scala/feral/lambda/ContextPlatform.scala index 77aa1118..5481a010 100644 --- a/lambda/jvm/src/main/scala/feral/lambda/ContextPlatform.scala +++ b/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) ) } diff --git a/lambda/jvm/src/main/scala/feral/lambda/IOLambdaPlatform.scala b/lambda/jvm/src/main/scala/feral/lambda/LambdaPlatform.scala similarity index 64% rename from lambda/jvm/src/main/scala/feral/lambda/IOLambdaPlatform.scala rename to lambda/jvm/src/main/scala/feral/lambda/LambdaPlatform.scala index 73a7fdcd..f9323f42 100644 --- a/lambda/jvm/src/main/scala/feral/lambda/IOLambdaPlatform.scala +++ b/lambda/jvm/src/main/scala/feral/lambda/LambdaPlatform.scala @@ -17,8 +17,8 @@ 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 @@ -26,35 +26,35 @@ 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) + } } diff --git a/lambda/shared/src/main/scala/feral/lambda/Context.scala b/lambda/shared/src/main/scala/feral/lambda/Context.scala index bbbccb6b..44ff28da 100644 --- a/lambda/shared/src/main/scala/feral/lambda/Context.scala +++ b/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 diff --git a/lambda/shared/src/main/scala/feral/lambda/IOLambda.scala b/lambda/shared/src/main/scala/feral/lambda/Lambda.scala similarity index 75% rename from lambda/shared/src/main/scala/feral/lambda/IOLambda.scala rename to lambda/shared/src/main/scala/feral/lambda/Lambda.scala index 81d22e81..4ffa2573 100644 --- a/lambda/shared/src/main/scala/feral/lambda/IOLambda.scala +++ b/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