From e6552f3b31b55396c652c196c5c3a9c3a6cfed71 Mon Sep 17 00:00:00 2001
From: Jakob Odersky <jakob@odersky.com>
Date: Tue, 31 Jul 2018 12:13:47 -0700
Subject: Add message bus and topic abstractions (#181)

---
 .../scala/xyz/driver/core/messaging/Topic.scala    | 43 ++++++++++++++++++++++
 1 file changed, 43 insertions(+)
 create mode 100644 src/main/scala/xyz/driver/core/messaging/Topic.scala

(limited to 'src/main/scala/xyz/driver/core/messaging/Topic.scala')

diff --git a/src/main/scala/xyz/driver/core/messaging/Topic.scala b/src/main/scala/xyz/driver/core/messaging/Topic.scala
new file mode 100644
index 0000000..32fd764
--- /dev/null
+++ b/src/main/scala/xyz/driver/core/messaging/Topic.scala
@@ -0,0 +1,43 @@
+package xyz.driver.core
+package messaging
+
+import java.nio.ByteBuffer
+
+/** A topic is a named group of messages that all share a common schema.
+  * @tparam Message type of messages sent over this topic */
+trait Topic[Message] {
+
+  /** Name of this topic (must be unique). */
+  def name: String
+
+  /** Convert a message to its wire format that will be sent over a bus. */
+  def serialize(message: Message): ByteBuffer
+
+  /** Convert a message from its wire format. */
+  def deserialize(message: ByteBuffer): Message
+
+}
+
+object Topic {
+
+  /** Create a new "raw" topic without a schema, providing access to the underlying bytes of messages. */
+  def raw(name0: String): Topic[ByteBuffer] = new Topic[ByteBuffer] {
+    def name                                                  = name0
+    override def serialize(message: ByteBuffer): ByteBuffer   = message
+    override def deserialize(message: ByteBuffer): ByteBuffer = message
+  }
+
+  /** Create a topic that represents data as UTF-8 encoded strings. */
+  def string(name0: String): Topic[String] = new Topic[String] {
+    def name = name0
+    override def serialize(message: String): ByteBuffer = {
+      ByteBuffer.wrap(message.getBytes("utf-8"))
+    }
+    override def deserialize(message: ByteBuffer): String = {
+      val bytes = new Array[Byte](message.remaining())
+      message.get(bytes)
+      new String(bytes, "utf-8")
+    }
+  }
+
+}
-- 
cgit v1.2.3