diff options
author | Jakob Odersky <jakob@odersky.com> | 2018-09-10 15:10:09 -0700 |
---|---|---|
committer | GitHub <noreply@github.com> | 2018-09-10 15:10:09 -0700 |
commit | f07bee564b11ee76fd065ec849a888bcf4e74e85 (patch) | |
tree | 538330c7592ccd4f555a07bf03aa91cace403be5 /src/main/scala/xyz/driver/core/messaging/CreateBeforeStream.scala | |
parent | 93ffc0a5c4b54a8beea10c3fd68e7a2d70c4c771 (diff) | |
download | driver-core-f07bee564b11ee76fd065ec849a888bcf4e74e85.tar.gz driver-core-f07bee564b11ee76fd065ec849a888bcf4e74e85.tar.bz2 driver-core-f07bee564b11ee76fd065ec849a888bcf4e74e85.zip |
Various message bus fixes (#212)v1.14.0
1. Move to pure mixin-based ("stackable traits") pattern.
2. Provide a "CreateOnDemand" mixin that ensures topics and
subscriptions have been created before they are used.
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()) - } - -} |