IOCloudEvents Function Trait for Google Cloud Functions

build.sbt

@@ -194,16 +194,23 @@ lazy val examples = crossProject(JSPlatform, JVMPlatform)
     )
   )
   .settings(commonSettings)
-  .dependsOn(lambda, lambdaHttp4s, googleCloudHttp4s)
+  .dependsOn(lambda, lambdaHttp4s, googleCloudHttp4s, googleCloud)
   .jsSettings(
     scalaJSUseMainModuleInitializer := true,
     Compile / mainClass := Some("feral.examples.http4sGoogleCloudHandler"),
     scalaJSLinkerConfig ~= { _.withModuleKind(ModuleKind.CommonJSModule) }
   )
   .jvmSettings(
     libraryDependencies ++= Seq(
-      "com.google.cloud.functions.invoker" % "java-function-invoker" % "1.4.3"
-    )
+      "com.google.cloud.functions.invoker" % "java-function-invoker" % "1.4.3",
+      "org.scalameta" %% "munit" % munitVersion % Test,
+      "org.typelevel" %% "munit-cats-effect" % munitCEVersion % Test,
+      "io.circe" %% "circe-literal" % circeVersion % Test,
+      "com.google.protobuf" % "protobuf-java" % "4.29.3" % Test,
+      "com.google.protobuf" % "protobuf-java-util" % "4.29.3" % Test,
+      "com.google.cloud" % "google-cloudevent-types" % "0.16.0" % Test
+    ),
+    Test / fork := true
   )
   .enablePlugins(NoPublishPlugin)
 
@@ -281,7 +288,12 @@ lazy val googleCloud = crossProject(JSPlatform, JVMPlatform)
       "org.typelevel" %%% "cats-effect" % catsEffectVersion,
       "org.scodec" %%% "scodec-bits" % "1.2.0",
       "io.circe" %%% "circe-core" % circeVersion,
+      "io.cloudevents" % "cloudevents-core" % "4.0.1",
       "io.circe" %%% "circe-generic" % circeVersion,
+      "io.circe" %% "circe-parser" % circeVersion,
+      "com.google.protobuf" % "protobuf-java" % "4.29.3",
+      "com.google.protobuf" % "protobuf-java-util" % "4.29.3",
+      "com.google.cloud" % "google-cloudevent-types" % "0.16.0",
       "org.scalameta" %%% "munit-scalacheck" % munitVersion % Test,
       "org.typelevel" %%% "munit-cats-effect" % munitCEVersion % Test,
       "io.circe" %%% "circe-literal" % circeVersion % Test

examples/jvm/src/main/scala/feral/examples/StackDriverLogging.scala

@@ -0,0 +1,55 @@
+/*
+ * Copyright 2021 Typelevel
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package feral.examples
+
+import cats.effect.IO
+import cats.effect.Resource
+import feral.googlecloud._
+import feral.lambda.INothing
+
+import java.util.Base64
+import java.util.logging.Logger
+
+import SubscribeToTopic._
+
+class StackDriverLogging extends IOCloudEventsFunction[PubSubBody, INothing] {
+
+  def handler: Resource[IO, ContextEventWithData[PubSubBody] => IO[Unit]] = {
+    val logger = Logger.getLogger(this.getClass.getName())
+
+    Resource.pure { event =>
+      val msg = event.data.getMessage.data
+
+      val er = msg.toBase64
+
+      if (!er.nonEmpty) {
+        logger.info("Hello World")
+        IO.unit
+      } else {
+        val decodedMessage = Base64.getDecoder().decode(er)
+        val result = new String(decodedMessage)
+
+        val output_message = s"Hello, $result"
+
+        IO.pure {
+          logger.info(s"data over the wire: ${output_message}")
+        } >> IO.println("done")
+      }
+
+    }
+  }
+}

examples/jvm/src/main/scala/feral/examples/SubscribeToTopic.scala

@@ -0,0 +1,67 @@
+/*
+ * Copyright 2021 Typelevel
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package feral.examples
+
+import cats.effect.IO
+import cats.effect.Resource
+import feral.googlecloud._
+import feral.googlecloud.events._
+import feral.lambda.INothing
+import io.circe.Decoder
+
+import java.util.logging.Logger
+
+import SubscribeToTopic._
+
+object SubscribeToTopic {
+
+  sealed abstract class PubSubBody {
+    def getMessage: PubsubMessage
+  }
+
+  object PubSubBody {
+    def apply(message: PubsubMessage): PubSubBody = new Impl(message)
+
+    implicit def decoder: Decoder[PubSubBody] = Decoder.forProduct1("message")(Impl.apply)
+
+    private case class Impl(
+        getMessage: PubsubMessage
+    ) extends PubSubBody {
+      override def productPrefix: String = "PubSubBody"
+    }
+
+  }
+
+}
+
+class SubscribeToTopic extends IOCloudEventsFunction[PubSubBody, INothing] {
+  val logger = Logger.getLogger(this.getClass.getName())
+
+  def handler: Resource[IO, ContextEventWithData[PubSubBody] => IO[Unit]] = {
+    Resource.pure { event =>
+      val msg = event.data.getMessage
+
+      IO.pure {
+        logger.info(s"Message ID: ${msg.messageId}")
+        logger.info(s"Publish Time: ${msg.publishTime}")
+        logger.info(s"Attributes: ${msg.attributes.mkString(", ")}")
+        logger.info(s"data: ${msg.data.toBase64}")
+      } >> IO.println("done")
+    }
+  }
+
+}

examples/jvm/src/test/scala/feral/examples/StackDriverLoggingTest.scala

@@ -0,0 +1,118 @@
+/*
+ * Copyright 2021 Typelevel
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package feral.examples
+
+import com.google.events.cloud.pubsub.v1.MessagePublishedData
+import com.google.events.cloud.pubsub.v1.PubsubMessage
+import com.google.protobuf.ByteString
+import com.google.protobuf.util.JsonFormat
+import com.google.protobuf.util.Timestamps
+import feral.examples.util.TestLogHandler
+import io.cloudevents.core.builder.CloudEventBuilder
+
+import java.util.logging.Logger
+
+class StackDriverLoggingTest extends munit.FunSuite {
+  val logger = Logger.getLogger("feral.examples.StackDriverLogging")
+  val testLogHandler = TestLogHandler()
+  override def beforeAll(): Unit = {
+    logger.setUseParentHandlers(false)
+    logger.addHandler(testLogHandler)
+  }
+
+  override def afterEach(context: AfterEach): Unit =
+    testLogHandler.clear()
+
+  test("Message sent over the wire should output \'hello name\' where name is Feral") {
+    val msg = "Feral"
+
+    // The message variable will spit out a protobuf format and converted to
+    // json format to produce the below
+    // "message": {
+    //     "attributes": {
+    //       "attr1": "attr1-value"
+    //     },
+    //     "data": "Feral",
+    //     "messageId": "message-id",
+    //     "publishTime": "2021-02-05T04:06:14.109Z"
+    //   }
+
+    val message = PubsubMessage
+      .newBuilder()
+      .setData(ByteString.copyFromUtf8(msg))
+      .putAttributes("attr1", "attr1-value")
+      .setMessageId("message-id")
+      .setPublishTime(Timestamps.parse("2021-02-05T04:06:14.109Z"))
+      .build()
+
+    val data = MessagePublishedData.newBuilder().setMessage(message).build()
+
+    val json_formatted_data = JsonFormat.printer().print(data)
+
+    val event = CloudEventBuilder
+      .v1()
+      .withId("1234-5678-9012-3456")
+      .withType("pubsub.message")
+      .withSource(java.net.URI.create("https://github.com/cloudevents/spec/pull/123"))
+      .withData(json_formatted_data.getBytes())
+      .build()
+
+    new StackDriverLogging().accept(event)
+
+    val messages = testLogHandler.getLog.map(_.getMessage())
+
+    assertEquals(
+      List("data over the wire: Hello, Feral"),
+      messages
+    )
+
+  }
+
+  test("No Message sent over the wire should output \'hello World\' default") {
+
+    val pubsub_message = PubsubMessage
+      .newBuilder()
+      .setData(ByteString.copyFromUtf8(""))
+      .putAttributes("attr1", "attr1-value")
+      .setMessageId("message-id")
+      .setPublishTime(Timestamps.parse("2021-02-05T04:06:14.109Z"))
+      .build()
+
+    val message_published_data =
+      MessagePublishedData.newBuilder().setMessage(pubsub_message).build()
+
+    val json_formatted_data =
+      JsonFormat.printer().alwaysPrintFieldsWithNoPresence().print(message_published_data)
+
+    val event = CloudEventBuilder
+      .v1()
+      .withId("1234-5678-9012-3456")
+      .withType("pubsub.message")
+      .withSource(java.net.URI.create("https://github.com/cloudevents/spec/pull/123"))
+      .withData(json_formatted_data.getBytes())
+      .build()
+
+    new StackDriverLogging().accept(event)
+
+    val messages = testLogHandler.getLog.map(_.getMessage())
+
+    assertEquals(
+      List("Hello World"),
+      messages
+    )
+  }
+}

examples/jvm/src/test/scala/feral/examples/SubscribeToTopicTest.scala

@@ -0,0 +1,88 @@
+/*
+ * Copyright 2021 Typelevel
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package feral.examples
+
+import feral.examples.util.TestLogHandler
+import io.circe.literal._
+import io.cloudevents.core.builder.CloudEventBuilder
+
+import java.util.Base64
+import java.util.logging.Logger
+
+class SubscribeToTopicTest extends munit.FunSuite {
+  val logger = Logger.getLogger("feral.examples.SubscribeToTopic")
+  val testLogHandler = TestLogHandler()
+  override def beforeAll(): Unit = {
+    logger.addHandler(testLogHandler)
+  }
+  // the test uses plain json and base 64 to construct the encodedData unlike Protobuf example in StackDriverLoggingTests
+  test("Google Cloud Function SubscribeToTopic should print pubsub message") {
+    val msg = "Hello World"
+    val encodedMessage = Base64.getEncoder().encodeToString(msg.getBytes())
+
+    val encodedData =
+      s"""{
+         |"message": {
+         |    "attributes": {
+         |      "attr1": "attr1-value"
+         |    },
+         |    "data": "$encodedMessage",
+         |    "messageId": "message-id",
+         |    "publishTime": "2021-02-05T04:06:14.109Z"
+         |  }
+         |}""".stripMargin
+
+    val event = CloudEventBuilder
+      .v1()
+      .withId("1234-5678-9012-3456")
+      .withType("pubsub.message")
+      .withSource(java.net.URI.create("https://github.com/cloudevents/spec/pull/123"))
+      .withData(encodedData.getBytes())
+      .build()
+
+    new SubscribeToTopic().accept(event)
+
+    val messages = testLogHandler.getLog
+
+    val first_mesage = messages
+      .filter(r => r.getMessage().contains("data"))
+      .head
+      .getMessage()
+      .split(":")(1)
+      .trim()
+
+    val res = json"""{
+      "message": {
+          "attributes": {
+              "attr1":"attr1-value"
+          },
+          "data": "SGVsbG8gV29ybGQ=",
+          "messageId": "message-id",
+          "publishTime":"2021-02-05T04:06:14.109Z"
+        }
+      }
+    """
+
+    val data =
+      res.hcursor.downField("message").downField("data").focus.flatMap(_.asString).get.trim()
+
+    assertEquals(
+      data,
+      first_mesage
+    )
+  }
+}