Custom Lambda Runtime

build.sbt

@@ -56,7 +56,7 @@ val Scala213 = "2.13.9"
 val Scala3 = "3.1.3"
 ThisBuild / crossScalaVersions := Seq(Scala212, Scala3, Scala213)
 
-val catsEffectVersion = "3.3.14"
+val catsEffectVersion = "3.4.0-RC1"
 val circeVersion = "0.14.3"
 val fs2Version = "3.3.0"
 val http4sVersion = "0.23.16"
@@ -121,8 +121,9 @@ lazy val lambdaRuntime = crossProject(JSPlatform, JVMPlatform, NativePlatform)
   .crossType(CrossType.Pure)
   .in(file("feral-lambda-runtime"))
   .settings(
-    name := "feral-lambda-kernel",
+    name := "feral-lambda-runtime",
     libraryDependencies ++= Seq(
+      "org.typelevel" %%% "cats-effect" % catsEffectVersion,
       "io.circe" %%% "circe-jawn" % circeVersion,
       "org.http4s" %%% "http4s-client" % http4sVersion,
       "org.http4s" %%% "http4s-circe" % http4sVersion

feral-lambda-runtime/src/main/scala/feral/lambda/runtime/LambdaRequest.scala

@@ -14,10 +14,8 @@ case class LambdaRequest(
     )
 
 object LambdaRequest {
-    // Still need to decide how to handle failed request or invalid header values
     def fromResponse[F[_]](response: Response[F])(implicit F: ApplicativeError[F, Throwable]): F[LambdaRequest] = {
-        F.pure(LambdaRequest(???))
-        //Unsure on best way to unpack the response headers into LambdaRequest case class
+        ???
     }
 
 }

feral-lambda-runtime/src/main/scala/feral/lambda/runtime/LambdaRuntime.scala

@@ -14,56 +14,54 @@
  * limitations under the License.
  */
 
-// TODO implement a runtime here
-// it should retrieve incoming events and handle them with the handler
-// it will run on a background fiber, whose lifecycle is controlled by the resource
-
 package feral.lambda
 package runtime
 
-import cats.Applicative
 import cats.syntax.all._
-import cats.effect.kernel.{Resource, Sync}
+import cats.effect.kernel.Sync
 import io.circe.Json
 import org.http4s.Method.POST
 import org.http4s.client.Client
 import org.http4s.circe.jsonEncoderWithPrinter
-
 import scala.concurrent.duration.FiniteDuration
 import java.util.concurrent.TimeUnit
 import org.http4s.{EntityEncoder, Uri}
 import org.http4s.client.dsl.Http4sClientDsl
 import io.circe._
-
-import java.time.Instant //safe to use in native?
 import cats.effect.kernel.Async
+import cats.effect.std.Env
+import org.http4s.implicits.http4sLiteralsSyntax
+
+// TODO apply function error handling
+  // How to run global or static code from handler?
+// TODO Custom AWS header models
+// TODO CognitoIdentity/ClientContext JSON encoding
 
 object FeralLambdaRuntime {
 
-  val LAMBDA_VERSION_DATE = "2018-06-01"
+  final val ApiVersion = "2018-06-01"
 
-  def apply[F[_]](client: Client[F])(handler: (Json, Context[F]) => F[Json])(implicit F: Async[F]): Resource[F, Unit] = {
-    F.background {
-      val runtimeUrl = getRuntimeUrl(LambdaReservedEnvVars.AWS_LAMBDA_RUNTIME_API)
-      implicit val jsonEncoder: EntityEncoder[F, Json] = jsonEncoderWithPrinter[F](Printer.noSpaces.copy(dropNullValues = true))
-      val http4sClientDsl = new Http4sClientDsl[F] {}
-      import http4sClientDsl._
-      (for {
-        request <- client.get(runtimeUrl)(LambdaRequest.fromResponse) // unsure how to deal with bad response
-        context <- createContext(request)
-        result <- handler(request.body, context)
-        invocationUrl = getInvocationUrl(LambdaReservedEnvVars.AWS_LAMBDA_RUNTIME_API, request.id)
-        _ <- client.successful(POST(result, invocationUrl))
-      } yield ()).foreverM
-    }
-  }.as(()) // how to handle Outcome error and cancellation?
+  def apply[F[_]](client: Client[F])(handler: (Json, Context[F]) => F[Json])(implicit F: Async[F]): F[Unit] = {
+    implicit val lambdaEnv: LambdaRuntimeEnv[F] = LambdaRuntimeEnv(Env.make) // maybe better way
+    implicit val jsonEncoder: EntityEncoder[F, Json] = jsonEncoderWithPrinter[F](Printer.noSpaces.copy(dropNullValues = true))
+    val http4sClientDsl = new Http4sClientDsl[F] {}
+    import http4sClientDsl._
+    (for {
+      runtimeApi <- lambdaEnv.lambdaRuntimeApi
+      request <- client.get(getRuntimeUrl(runtimeApi))(LambdaRequest.fromResponse)
+      context <- createContext(request)
+      result <- handler(request.body, context)
+      invocationUrl = getInvocationUrl(LambdaRuntimeEnv.AWS_LAMBDA_RUNTIME_API, request.id)
+      _ <- client.successful(POST(result, invocationUrl))
+    } yield ()).foreverM
+  }.void
 
-  private def createContext[F[_]](request: LambdaRequest)(implicit F: Sync[F]): F[Context[F]] = for {
-    functionName <- envVar(LambdaReservedEnvVars.AWS_LAMBDA_FUNCTION_NAME)
-    functionVersion <- envVar(LambdaReservedEnvVars.AWS_LAMBDA_FUNCTION_VERSION)
-    functionMemorySize <- envVar(LambdaReservedEnvVars.AWS_LAMBDA_FUNCTION_MEMORY_SIZE).map(_.toInt)
-    logGroupName <- envVar(LambdaReservedEnvVars.AWS_LAMBDA_LOG_GROUP_NAME)
-    logStreamName <- envVar(LambdaReservedEnvVars.AWS_LAMBDA_LOG_STREAM_NAME)
+  private def createContext[F[_]](request: LambdaRequest)(implicit F: Sync[F], lambdaEnv: LambdaRuntimeEnv[F]): F[Context[F]] = for {
+    functionName <- lambdaEnv.lambdaFunctionName
+    functionVersion <- lambdaEnv.lambdaFunctionVersion
+    functionMemorySize <- lambdaEnv.lambdaFunctionMemorySize
+    logGroupName <- lambdaEnv.lambdaLogGroupName
+    logStreamName <- lambdaEnv.lambdaLogStreamName
   } yield {
     new Context[F](
       functionName,
@@ -73,23 +71,18 @@ object FeralLambdaRuntime {
       request.id,
       logGroupName,
       logStreamName,
-      None, //need
-      None, //need
-      F.delay(FiniteDuration(request.deadlineTimeInMs.toEpochMilli - Instant.now.toEpochMilli, TimeUnit.MILLISECONDS))
+      None,
+      None,
+      F.realTimeInstant.map(curTime => FiniteDuration(request.deadlineTimeInMs.toEpochMilli - curTime.toEpochMilli, TimeUnit.MILLISECONDS))/// how to provide test version?, maybe separate Clock parameter?
     )
   }
 
-  private def getRuntimeUrl(api: String) = Uri.unsafeFromString(s"http://$api/$LAMBDA_VERSION_DATE/runtime/invocation/next") //need to be unsafe?
+  private def getRuntimeUrl(api: String) = uri"http://$api/$ApiVersion/runtime/invocation/next"
 
-  private def getInvocationUrl(api: String, id: String) = Uri.unsafeFromString(s"http://$api/$LAMBDA_VERSION_DATE/runtime/invocation/$id/response")
+  private def getInvocationUrl(api: String, id: String) = uri"http://$api/$ApiVersion/runtime/invocation/$id/response"
 
-  // Called if initialization of handler function fails, seems impossible here since handler function is provided as constructor?
-  private def getInitErrorUrl(api: String) = Uri.unsafeFromString(s"http://$api/$LAMBDA_VERSION_DATE/runtime/init/error")
+  private def getInitErrorUrl(api: String) = uri"http://$api/$ApiVersion/runtime/init/error"
 
-  // from docs, called "if the function returns an error or the runtime encounters an error", will be used after error handling implemented
-  private def getInvocationErrorUrl(api: String, errorType: String) = Uri.unsafeFromString(s"http://$api/$LAMBDA_VERSION_DATE/runtime/invocation/$errorType/error")
+  private def getInvocationErrorUrl(api: String, errorType: String) = uri"http://$api/$ApiVersion/runtime/invocation/$errorType/error"
 
-  private def envVar[F[_]](envVar: String)(implicit F: Sync[F]): F[String] = {
-    F.delay(sys.env(envVar))
-  }
 }

feral-lambda-runtime/src/main/scala/feral/lambda/runtime/LambdaRuntimeEnv.scala

@@ -0,0 +1,70 @@
+package feral.lambda.runtime
+
+import cats.{ApplicativeError, Functor}
+import cats.effect.kernel.Sync
+import cats.effect.std.Env
+import cats.syntax.all._
+
+trait LambdaRuntimeEnv[F[_]] {
+  def handler: F[String]
+  def region: F[String]
+  def executionEnv: F[String]
+  def lambdaFunctionName: F[String]
+  def lambdaFunctionMemorySize: F[Int]
+  def lambdaFunctionVersion: F[String]
+  def lambdaLogGroupName: F[String]
+  def lambdaLogStreamName: F[String]
+  def accessKeyId: F[String]
+  def secretAccessKey: F[String]
+  def lambdaRuntimeApi: F[String]
+  def lambdaTaskRoot: F[String]
+  def lambdaRuntimeDir: F[String]
+  def timezone: F[String]
+}
+
+object LambdaRuntimeEnv {
+  final val HANDLER = "_HANDLER"
+  final val AWS_REGION = "AWS_REGION"
+  final val AWS_EXECUTION_ENV = "AWS_EXECUTION_ENV"
+  final val AWS_LAMBDA_FUNCTION_NAME = "AWS_LAMBDA_FUNCTION_NAME"
+  final val AWS_LAMBDA_FUNCTION_MEMORY_SIZE = "AWS_LAMBDA_FUNCTION_MEMORY_SIZE"
+  final val AWS_LAMBDA_FUNCTION_VERSION = "AWS_LAMBDA_FUNCTION_VERSION"
+  final val AWS_LAMBDA_LOG_GROUP_NAME = "AWS_LAMBDA_LOG_GROUP_NAME"
+  final val AWS_LAMBDA_LOG_STREAM_NAME = "AWS_LAMBDA_LOG_STREAM_NAME"
+  final val AWS_ACCESS_KEY_ID = "AWS_ACCESS_KEY_ID"
+  final val AWS_SECRET_ACCESS_KEY = "AWS_SECRET_ACCESS_KEY"
+  final val AWS_LAMBDA_RUNTIME_API = "AWS_LAMBDA_RUNTIME_API"
+  final val LAMBDA_TASK_ROOT = "LAMBDA_TASK_ROOT"
+  final val LAMBDA_RUNTIME_DIR = "LAMBDA_RUNTIME_DIR"
+  final val TZ = "TZ"
+
+  def apply[F[_]: Functor](env: Env[F]): LambdaRuntimeEnv[F] = new LambdaRuntimeEnv[F] { // should be error effect?
+    override def handler: F[String] = env.get(HANDLER).map(_.get)
+
+    override def region: F[String] = env.get(AWS_REGION).map(_.get)
+
+    override def executionEnv: F[String] = env.get(AWS_EXECUTION_ENV).map(_.get)
+
+    override def lambdaFunctionName: F[String] = env.get(AWS_LAMBDA_FUNCTION_NAME).map(_.get)
+
+    override def lambdaFunctionMemorySize: F[Int] = env.get(AWS_LAMBDA_FUNCTION_MEMORY_SIZE).map(_.get)
+
+    override def lambdaFunctionVersion: F[String] = env.get(AWS_LAMBDA_FUNCTION_VERSION).map(_.get)
+
+    override def lambdaLogGroupName: F[String] = env.get(AWS_LAMBDA_LOG_GROUP_NAME).map(_.get)
+
+    override def lambdaLogStreamName: F[String] = env.get(AWS_LAMBDA_LOG_STREAM_NAME).map(_.get)
+
+    override def accessKeyId: F[String] = env.get(AWS_ACCESS_KEY_ID).map(_.get)
+
+    override def secretAccessKey: F[String] = env.get(AWS_SECRET_ACCESS_KEY).map(_.get)
+
+    override def lambdaRuntimeApi: F[String] = env.get(AWS_LAMBDA_RUNTIME_API).map(_.get)
+
+    override def lambdaTaskRoot: F[String] = env.get(LAMBDA_TASK_ROOT).map(_.get)
+
+    override def lambdaRuntimeDir: F[String] = env.get(LAMBDA_RUNTIME_DIR).map(_.get)
+
+    override def timezone: F[String] = env.get(TZ).map(_.get)
+  }
+}