aboutsummaryrefslogtreecommitdiff
path: root/src/main/scala/xyz/driver/core/messaging/CreateBeforeStream.scala
diff options
context:
space:
mode:
authorJakob Odersky <jakob@odersky.com>2018-07-31 12:13:47 -0700
committerGitHub <noreply@github.com>2018-07-31 12:13:47 -0700
commite6552f3b31b55396c652c196c5c3a9c3a6cfed71 (patch)
treef6826eac8bff8470683547006d1e64d2bc425d55 /src/main/scala/xyz/driver/core/messaging/CreateBeforeStream.scala
parentdb0c9bebee4cbc587d4da0a624f671ffcf7a649f (diff)
downloaddriver-core-e6552f3b31b55396c652c196c5c3a9c3a6cfed71.tar.gz
driver-core-e6552f3b31b55396c652c196c5c3a9c3a6cfed71.tar.bz2
driver-core-e6552f3b31b55396c652c196c5c3a9c3a6cfed71.zip
Add message bus and topic abstractions (#181)v1.12.0
Diffstat (limited to 'src/main/scala/xyz/driver/core/messaging/CreateBeforeStream.scala')
-rw-r--r--src/main/scala/xyz/driver/core/messaging/CreateBeforeStream.scala32
1 files changed, 32 insertions, 0 deletions
diff --git a/src/main/scala/xyz/driver/core/messaging/CreateBeforeStream.scala b/src/main/scala/xyz/driver/core/messaging/CreateBeforeStream.scala
new file mode 100644
index 0000000..bd8e422
--- /dev/null
+++ b/src/main/scala/xyz/driver/core/messaging/CreateBeforeStream.scala
@@ -0,0 +1,32 @@
+package xyz.driver.core
+package messaging
+
+import akka.stream.scaladsl.{Flow, Source}
+
+import scala.async.Async.{async, await}
+import scala.concurrent.Future
+
+/** Utility mixin that will ensure that topics and subscriptions exist before attempting to stream from or to them. */
+trait CreateBeforeStream extends StreamBus {
+
+ def createTopic(topic: Topic[_]): Future[Unit]
+ def createSubscription(topic: Topic[_], config: SubscriptionConfig): Future[Unit]
+
+ override def publish[A](topic: Topic[A]): Flow[A, A, _] = {
+ def create(): Future[Flow[A, A, _]] = async {
+ await(createTopic(topic))
+ super.publish(topic)
+ }
+ Flow.lazyInitAsync[A, A, Any](() => create())
+ }
+
+ override def subscribe[A](topic: Topic[A], config: SubscriptionConfig): Source[Message[A], _] = {
+ def create(): Future[Source[Message[A], _]] = async {
+ await(createTopic(topic))
+ await(createSubscription(topic, config))
+ super.subscribe(topic, config)
+ }
+ Source.fromFutureSource[Message[A], Any](create())
+ }
+
+}