Skip to content

Commit e75cd91

Browse files
Yummy-YumsYummy-Yums
committed
iocloudevents trait with gcp examples
1 parent 2aed831 commit e75cd91

7 files changed

Lines changed: 639 additions & 0 deletions

File tree

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

Lines changed: 40 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,40 @@
1+
package feral.examples
2+
3+
4+
import cats.effect.IO
5+
import cats.effect.Resource
6+
import feral.googlecloud._
7+
import feral.lambda.INothing
8+
9+
import java.util.Base64
10+
import java.util.logging.Logger
11+
12+
import SubscribeToTopic._
13+
14+
class StackDriverLogging extends IOCloudEventsFunction[PubSubBody, INothing]{
15+
16+
def handler: Resource[IO, ContextEventWithData[PubSubBody] => IO[Unit]] = {
17+
val logger = Logger.getLogger(this.getClass.getName())
18+
19+
Resource.pure { event =>
20+
val msg = event.data.getMessage.data
21+
22+
val er = msg.toBase64
23+
24+
if (!er.nonEmpty) {
25+
logger.info("Hello World")
26+
IO.unit
27+
} else {
28+
val decodedMessage = Base64.getDecoder().decode(er)
29+
val result = new String(decodedMessage)
30+
31+
val output_message = s"Hello, $result"
32+
33+
IO.pure {
34+
logger.info(s"data over the wire: ${output_message}")
35+
} >> IO.println("done")
36+
}
37+
38+
}
39+
}
40+
}

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

Lines changed: 50 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,50 @@
1+
package feral.examples
2+
3+
import cats.effect.IO
4+
import cats.effect.Resource
5+
import feral.googlecloud._
6+
import feral.googlecloud.events._
7+
import feral.lambda.INothing
8+
import io.circe.Decoder
9+
10+
import java.util.logging.Logger
11+
12+
import SubscribeToTopic._
13+
14+
object SubscribeToTopic {
15+
16+
sealed abstract class PubSubBody {
17+
def getMessage: PubsubMessage
18+
}
19+
20+
object PubSubBody {
21+
def apply(message: PubsubMessage): PubSubBody = new Impl(message)
22+
23+
implicit def decoder: Decoder[PubSubBody] = Decoder.forProduct1("message")(Impl.apply)
24+
25+
private case class Impl(
26+
getMessage: PubsubMessage
27+
) extends PubSubBody {
28+
override def productPrefix: String = "PubSubBody"
29+
}
30+
31+
}
32+
33+
}
34+
35+
class SubscribeToTopic extends IOCloudEventsFunction[PubSubBody, INothing]{
36+
val logger = Logger.getLogger(this.getClass.getName())
37+
38+
def handler: Resource[IO, ContextEventWithData[PubSubBody] => IO[Unit]] = {
39+
Resource.pure { event =>
40+
val msg = event.data.getMessage
41+
IO.pure {
42+
logger.info(s"Message ID: ${msg.messageId}")
43+
logger.info(s"Publish Time: ${msg.publishTime}")
44+
logger.info(s"Attributes: ${msg.attributes.mkString(", ")}")
45+
logger.info(s"data: ${msg.data}")
46+
} >> IO.println("done")
47+
}
48+
}
49+
50+
}

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

Lines changed: 101 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,101 @@
1+
package feral.examples
2+
3+
import io.cloudevents.core.builder.CloudEventBuilder
4+
import com.google.events.cloud.pubsub.v1.PubsubMessage
5+
import com.google.events.cloud.pubsub.v1.MessagePublishedData
6+
import com.google.protobuf.{ByteString}
7+
import com.google.protobuf.util.{Timestamps, JsonFormat}
8+
9+
import java.util.logging.Logger
10+
11+
12+
class StackDriverLoggingTest extends munit.FunSuite {
13+
val logger = Logger.getLogger("feral.examples.StackDriverLogging")
14+
val testLogHandler = TestLogHandler()
15+
override def beforeAll(): Unit = {
16+
logger.setUseParentHandlers(false)
17+
logger.addHandler(testLogHandler)
18+
}
19+
20+
override def afterEach(context: AfterEach): Unit =
21+
testLogHandler.clear()
22+
23+
test("Message sent over the wire should output \'hello name\' where name is Feral") {
24+
val msg = "Feral"
25+
26+
// The message variable will spit out a protobuf format and converted to
27+
// json format to produce the below
28+
// "message": {
29+
// "attributes": {
30+
// "attr1": "attr1-value"
31+
// },
32+
// "data": "Feral",
33+
// "messageId": "message-id",
34+
// "publishTime": "2021-02-05T04:06:14.109Z"
35+
// }
36+
37+
val message = PubsubMessage.newBuilder()
38+
.setData(ByteString.copyFromUtf8(msg))
39+
.putAttributes("attr1", "attr1-value")
40+
.setMessageId("message-id")
41+
.setPublishTime(Timestamps.parse("2021-02-05T04:06:14.109Z"))
42+
.build()
43+
44+
val data = MessagePublishedData.newBuilder()
45+
.setMessage(message)
46+
.build()
47+
48+
val json_formatted_data = JsonFormat.printer().print(data)
49+
50+
val event = CloudEventBuilder.v1()
51+
.withId("1234-5678-9012-3456")
52+
.withType("pubsub.message")
53+
.withSource(java.net.URI.create("https://github.com/cloudevents/spec/pull/123"))
54+
.withData(json_formatted_data.getBytes())
55+
.build()
56+
57+
new StackDriverLogging().accept(event)
58+
59+
val messages = testLogHandler.getLog.map(_.getMessage())
60+
61+
assertEquals(
62+
List("data over the wire: Hello, Feral"),
63+
messages
64+
)
65+
66+
}
67+
68+
test("No Message sent over the wire should output \'hello World\' default") {
69+
70+
val pubsub_message = PubsubMessage.newBuilder()
71+
.setData(ByteString.copyFromUtf8(""))
72+
.putAttributes("attr1", "attr1-value")
73+
.setMessageId("message-id")
74+
.setPublishTime(Timestamps.parse("2021-02-05T04:06:14.109Z"))
75+
.build()
76+
77+
val message_published_data = MessagePublishedData.newBuilder()
78+
.setMessage(pubsub_message)
79+
.build()
80+
81+
val json_formatted_data = JsonFormat.printer()
82+
.alwaysPrintFieldsWithNoPresence()
83+
.print(message_published_data)
84+
85+
val event = CloudEventBuilder.v1()
86+
.withId("1234-5678-9012-3456")
87+
.withType("pubsub.message")
88+
.withSource(java.net.URI.create("https://github.com/cloudevents/spec/pull/123"))
89+
.withData(json_formatted_data.getBytes())
90+
.build()
91+
92+
new StackDriverLogging().accept(event)
93+
94+
val messages = testLogHandler.getLog.map(_.getMessage())
95+
96+
assertEquals(
97+
List("Hello World"),
98+
messages
99+
)
100+
}
101+
}

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

Lines changed: 93 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,93 @@
1+
package feral.examples
2+
3+
import io.circe.literal._
4+
import io.cloudevents.core.builder.CloudEventBuilder
5+
6+
import java.util.Base64
7+
import java.util.logging.Handler
8+
import java.util.logging.Level
9+
import java.util.logging.LogRecord
10+
import java.util.logging.Logger
11+
12+
class SubscribeToTopicTest extends munit.FunSuite {
13+
val logger = Logger.getLogger("feral.examples.SubscribeToTopic")
14+
val testLogHandler = TestLogHandler()
15+
override def beforeAll(): Unit = {
16+
logger.addHandler(testLogHandler)
17+
}
18+
// the test uses plain json and base 64 to construct the encodedData unlike Protobuf example in StackDriverLoggingTests
19+
test("Google Cloud Function SubscribeToTopic should print pubsub message") {
20+
val msg = "Hello World"
21+
val encodedMessage = Base64.getEncoder().encodeToString(msg.getBytes())
22+
23+
val encodedData =
24+
s"""{
25+
|"message": {
26+
| "attributes": {
27+
| "attr1": "attr1-value"
28+
| },
29+
| "data": "$encodedMessage",
30+
| "messageId": "message-id",
31+
| "publishTime": "2021-02-05T04:06:14.109Z"
32+
| }
33+
|}""".stripMargin
34+
35+
val event = CloudEventBuilder.v1()
36+
.withId("1234-5678-9012-3456")
37+
.withType("pubsub.message")
38+
.withSource(java.net.URI.create("https://github.com/cloudevents/spec/pull/123"))
39+
.withData(encodedData.getBytes())
40+
.build()
41+
42+
new SubscribeToTopic().accept(event)
43+
44+
val messages = testLogHandler.getLog
45+
46+
val first_mesage = messages
47+
.filter(r => r.getMessage().contains("data"))
48+
.head
49+
.getMessage()
50+
.split(":")(1)
51+
.trim()
52+
53+
val res = json"""{
54+
"message": {
55+
"attributes": {
56+
"attr1":"attr1-value"
57+
},
58+
"data": "SGVsbG8gV29ybGQ=",
59+
"messageId": "message-id",
60+
"publishTime":"2021-02-05T04:06:14.109Z"
61+
}
62+
}
63+
"""
64+
65+
val data = res.hcursor
66+
.downField("message")
67+
.downField("data")
68+
.focus
69+
.flatMap(_.asString)
70+
.get.trim()
71+
72+
assertEquals(
73+
data,
74+
first_mesage
75+
)
76+
}
77+
}
78+
79+
class TestLogHandler extends Handler {
80+
81+
setLevel(Level.ALL)
82+
private val log_store: collection.mutable.ListBuffer[LogRecord] = scala.collection.mutable.ListBuffer.empty
83+
84+
def getLog = log_store.result()
85+
def clear(): Unit = log_store.clear()
86+
def close(): Unit = ()
87+
def flush(): Unit = ()
88+
def publish(record: LogRecord): Unit = log_store += record
89+
}
90+
91+
object TestLogHandler {
92+
def apply(): TestLogHandler = new TestLogHandler()
93+
}

‎google-cloud/jvm/src/main/scala/feral/googlecloud/Context.scala‎

Lines changed: 82 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,82 @@
1+
/*
2+
* Copyright 2021 Typelevel
3+
*
4+
* Licensed under the Apache License, Version 2.0 (the "License");
5+
* you may not use this file except in compliance with the License.
6+
* You may obtain a copy of the License at
7+
*
8+
* http://www.apache.org/licenses/LICENSE-2.0
9+
*
10+
* Unless required by applicable law or agreed to in writing, software
11+
* distributed under the License is distributed on an "AS IS" BASIS,
12+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13+
* See the License for the specific language governing permissions and
14+
* limitations under the License.
15+
*/
16+
17+
package feral.googlecloud
18+
19+
import io.circe.Decoder
20+
import io.circe.DecodingFailure
21+
import io.circe._
22+
import io.circe.parser._
23+
import io.cloudevents.CloudEvent
24+
25+
import java.net.URI
26+
import java.nio.charset.StandardCharsets
27+
import java.time.OffsetDateTime
28+
import scala.jdk.CollectionConverters._
29+
30+
final case class CloudEventContext(
31+
id: String,
32+
`type`: String,
33+
source: URI,
34+
dataContentType: Option[String],
35+
dataSchema: Option[URI],
36+
subject: Option[String],
37+
time: Option[OffsetDateTime],
38+
extensions: Map[String, String]
39+
)
40+
41+
object CloudEventContext {
42+
def from(event: CloudEvent): CloudEventContext = {
43+
CloudEventContext(
44+
id = event.getId(),
45+
`type` = event.getType(),
46+
source = event.getSource(),
47+
dataContentType = Option(event.getDataContentType()),
48+
dataSchema = Option(event.getDataSchema()),
49+
subject = Option(event.getSubject()),
50+
time = Option(event.getTime()),
51+
extensions = event.getAttributeNames
52+
.asScala
53+
.filterNot(Set("id","type","source","time","datacontenttype","subject"))
54+
.map(k => k -> event.getAttribute(k).toString)
55+
.toMap
56+
)
57+
}
58+
}
59+
60+
final case class ContextEventWithData[A](
61+
context: CloudEventContext,
62+
data: A
63+
)
64+
65+
66+
object ContextEventWithData {
67+
68+
def from[A: Decoder](event: CloudEvent): Either[Error, ContextEventWithData[A]] = {
69+
val context = CloudEventContext.from(event)
70+
71+
Option(event.getData())
72+
.toRight(DecodingFailure("CloudEvent has no data", Nil))
73+
.flatMap{ data =>
74+
//TODO - maybe refactor this to only use String instead of StringBuilder?
75+
val res = new StringBuilder()
76+
res.append(new String(data.toBytes(), StandardCharsets.UTF_8))
77+
parse(res.result()).flatMap(_.as[A])
78+
}
79+
.map(ContextEventWithData(context, _))
80+
}
81+
82+
}

0 commit comments

Comments
 (0)