raboof/pekko
1
0
Fork 0
Code Issues Pull requests Projects Releases 6 Packages Wiki Activity Actions 11

Allow reuse of journal loading logic (#24990)

Browse source 
* Allow reuse of journal loading logic

Probably can be generialised even more. At present i've tested this with
the akka persistence updater extension.

* Make the implicit explicit

* Added mima filter for removing internal class

* Review feedback
This commit is contained in:
Christopher Batey 2018-04-26 14:00:15 +01:00 • committed by Konrad `ktoso` Malawski
parent 90c2ce9f13
commit 9051e2fcda
5 changed files with 121 additions and 70 deletions
Download patch file Download diff file Expand all files Collapse all files

86
akka-persistence-query/src/main/scala/akka/persistence/query/PersistenceQuery.scala
View file

@ -4,15 +4,14 @@
package akka.persistence.query
import java.util.concurrent.atomic.AtomicReference
import akka.actor._
import akka.event.Logging
import scala.annotation.tailrec
import scala.util.Failure
import akka.annotation.InternalApi
import akka.persistence.query.scaladsl.ReadJournal
import akka.persistence.{ PersistencePlugin, PluginProvider }
import com.typesafe.config.{ Config, ConfigFactory }
import scala.reflect.ClassTag
/**
* Persistence extension for queries.
*/
@ -24,23 +23,20 @@ object PersistenceQuery extends ExtensionId[PersistenceQuery] with ExtensionIdPr
def createExtension(system: ExtendedActorSystem): PersistenceQuery = new PersistenceQuery(system)
def lookup() = PersistenceQuery
def lookup(): PersistenceQuery.type = PersistenceQuery
/** INTERNAL API. */
private[persistence] case class PluginHolder(
scaladslPlugin: scaladsl.ReadJournal, javadslPlugin: akka.persistence.query.javadsl.ReadJournal)
extends Extension
@InternalApi
private[akka] val pluginProvider: PluginProvider[ReadJournalProvider, ReadJournal, javadsl.ReadJournal] =
new PluginProvider[ReadJournalProvider, scaladsl.ReadJournal, javadsl.ReadJournal] {
override def scalaDsl(t: ReadJournalProvider): ReadJournal = t.scaladslReadJournal()
override def javaDsl(t: ReadJournalProvider): javadsl.ReadJournal = t.javadslReadJournal()
}
}
class PersistenceQuery(system: ExtendedActorSystem) extends Extension {
import PersistenceQuery._
private val log = Logging(system, getClass)
/** Discovered query plugins. */
private val readJournalPluginExtensionIds = new AtomicReference[Map[String, ExtensionId[PluginHolder]]](Map.empty)
class PersistenceQuery(system: ExtendedActorSystem)
extends PersistencePlugin[scaladsl.ReadJournal, javadsl.ReadJournal, ReadJournalProvider](system)(ClassTag(classOf[ReadJournalProvider]), PersistenceQuery.pluginProvider)
with Extension {
/**
* Scala API: Returns the [[akka.persistence.query.scaladsl.ReadJournal]] specified by the given
* read journal configuration entry.
@ -49,7 +45,7 @@ class PersistenceQuery(system: ExtendedActorSystem) extends Extension {
* config.
*/
final def readJournalFor[T <: scaladsl.ReadJournal](readJournalPluginId: String, readJournalPluginConfig: Config): T =
readJournalPluginFor(readJournalPluginId, readJournalPluginConfig).scaladslPlugin.asInstanceOf[T]
pluginFor(readJournalPluginId, readJournalPluginConfig).scaladslPlugin.asInstanceOf[T]
/**
* Scala API: Returns the [[akka.persistence.query.scaladsl.ReadJournal]] specified by the given
@ -63,57 +59,9 @@ class PersistenceQuery(system: ExtendedActorSystem) extends Extension {
* read journal configuration entry.
*/
final def getReadJournalFor[T <: javadsl.ReadJournal](clazz: Class[T], readJournalPluginId: String, readJournalPluginConfig: Config): T =
readJournalPluginFor(readJournalPluginId, readJournalPluginConfig).javadslPlugin.asInstanceOf[T]
pluginFor(readJournalPluginId, readJournalPluginConfig).javadslPlugin.asInstanceOf[T]
final def getReadJournalFor[T <: javadsl.ReadJournal](clazz: Class[T], readJournalPluginId: String): T = getReadJournalFor[T](clazz, readJournalPluginId, ConfigFactory.empty())
@tailrec private def readJournalPluginFor(readJournalPluginId: String, readJournalPluginConfig: Config): PluginHolder = {
val configPath = readJournalPluginId
val extensionIdMap = readJournalPluginExtensionIds.get
extensionIdMap.get(configPath) match {
case Some(extensionId) ⇒
extensionId(system)
case None ⇒
val extensionId = new ExtensionId[PluginHolder] {
override def createExtension(system: ExtendedActorSystem): PluginHolder = {
val provider = createPlugin(configPath, readJournalPluginConfig)
PluginHolder(provider.scaladslReadJournal(), provider.javadslReadJournal())
}
}
readJournalPluginExtensionIds.compareAndSet(extensionIdMap, extensionIdMap.updated(configPath, extensionId))
readJournalPluginFor(readJournalPluginId, readJournalPluginConfig) // Recursive invocation.
}
}
private def createPlugin(configPath: String, readJournalPluginConfig: Config): ReadJournalProvider = {
val mergedConfig = readJournalPluginConfig.withFallback(system.settings.config)
require(
!isEmpty(configPath) && mergedConfig.hasPath(configPath),
s"'reference.conf' is missing persistence read journal plugin config path: '${configPath}'")
val pluginConfig = mergedConfig.getConfig(configPath)
val pluginClassName = pluginConfig.getString("class")
log.debug(s"Create plugin: ${configPath} ${pluginClassName}")
val pluginClass = system.dynamicAccess.getClassFor[AnyRef](pluginClassName).get
def instantiate(args: collection.immutable.Seq[(Class[_], AnyRef)]) =
system.dynamicAccess.createInstanceFor[ReadJournalProvider](pluginClass, args)
instantiate((classOf[ExtendedActorSystem], system) :: (classOf[Config], pluginConfig) ::
(classOf[String], configPath) :: Nil)
.recoverWith {
case x: NoSuchMethodException ⇒ instantiate(
(classOf[ExtendedActorSystem], system) :: (classOf[Config], pluginConfig) :: Nil)
}
.recoverWith { case x: NoSuchMethodException ⇒ instantiate((classOf[ExtendedActorSystem], system) :: Nil) }
.recoverWith { case x: NoSuchMethodException ⇒ instantiate(Nil) }
.recoverWith {
case ex: Exception ⇒ Failure.apply(
new IllegalArgumentException("Unable to create read journal plugin instance for path " +
s"[$configPath], class [$pluginClassName]!", ex))
}.get
}
/** Check for default or missing identity. */
private def isEmpty(text: String) = text == null || text.length == 0
}