diff options
Diffstat (limited to 'src/main/scala/xyz/driver/core/messaging/CreateBeforeStream.scala')
-rw-r--r-- | src/main/scala/xyz/driver/core/messaging/CreateBeforeStream.scala | 32 |
1 files changed, 0 insertions, 32 deletions
diff --git a/src/main/scala/xyz/driver/core/messaging/CreateBeforeStream.scala b/src/main/scala/xyz/driver/core/messaging/CreateBeforeStream.scala deleted file mode 100644 index bd8e422..0000000 --- a/src/main/scala/xyz/driver/core/messaging/CreateBeforeStream.scala +++ /dev/null @@ -1,32 +0,0 @@ -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()) - } - -} |