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/CreateOnDemand.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/CreateOnDemand.scala')
-rw-r--r-- | src/main/scala/xyz/driver/core/messaging/CreateOnDemand.scala | 49 |
1 files changed, 49 insertions, 0 deletions
diff --git a/src/main/scala/xyz/driver/core/messaging/CreateOnDemand.scala b/src/main/scala/xyz/driver/core/messaging/CreateOnDemand.scala new file mode 100644 index 0000000..1af5308 --- /dev/null +++ b/src/main/scala/xyz/driver/core/messaging/CreateOnDemand.scala @@ -0,0 +1,49 @@ +package xyz.driver.core +package messaging + +import java.util.concurrent.ConcurrentHashMap + +import scala.async.Async.{async, await} +import scala.concurrent.Future + +/** Utility mixin that will ensure that topics and subscriptions exist before + * attempting to read or write from or to them. + */ +trait CreateOnDemand extends Bus { + + /** Create the given topic. This operation is idempotent and is expected to succeed if the topic + * already exists. + */ + def createTopic(topic: Topic[_]): Future[Unit] + + /** Create the given subscription. This operation is idempotent and is expected to succeed if the subscription + * already exists. + */ + def createSubscription(topic: Topic[_], config: SubscriptionConfig): Future[Unit] + + private val createdTopics = new ConcurrentHashMap[Topic[_], Future[Unit]] + private val createdSubscriptions = new ConcurrentHashMap[(Topic[_], SubscriptionConfig), Future[Unit]] + + private def ensureTopic(topic: Topic[_]) = + createdTopics.computeIfAbsent(topic, t => createTopic(t)) + + private def ensureSubscription(topic: Topic[_], config: SubscriptionConfig) = + createdSubscriptions.computeIfAbsent(topic -> config, { + case (t, c) => createSubscription(t, c) + }) + + abstract override def publishMessages[A](topic: Topic[A], messages: Seq[A]): Future[Unit] = async { + await(ensureTopic(topic)) + await(super.publishMessages(topic, messages)) + } + + abstract override def fetchMessages[A]( + topic: Topic[A], + config: SubscriptionConfig, + maxMessages: Int): Future[Seq[Message[A]]] = async { + await(ensureTopic(topic)) + await(ensureSubscription(topic, config)) + await(super.fetchMessages(topic, config, maxMessages)) + } + +} |