d0bc8a6400
Further changes - remove obsolete identity checks in Eventsourced - fix wrong serialize-messages config in tests
128 lines
4 KiB
Scala
128 lines
4 KiB
Scala
/**
|
|
* Copyright (C) 2009-2013 Typesafe Inc. <http://www.typesafe.com>
|
|
*/
|
|
|
|
package docs.persistence
|
|
|
|
//#plugin-imports
|
|
import scala.concurrent.Future
|
|
import scala.collection.immutable.Seq
|
|
//#plugin-imports
|
|
|
|
import com.typesafe.config._
|
|
|
|
import org.scalatest.WordSpec
|
|
|
|
import akka.actor.ActorSystem
|
|
//#plugin-imports
|
|
import akka.persistence._
|
|
import akka.persistence.journal._
|
|
import akka.persistence.snapshot._
|
|
//#plugin-imports
|
|
|
|
object PersistencePluginDocSpec {
|
|
val config =
|
|
"""
|
|
//#max-batch-size
|
|
akka.persistence.journal.max-batch-size = 200
|
|
//#max-batch-size
|
|
//#journal-config
|
|
akka.persistence.journal.leveldb.dir = "target/journal"
|
|
//#journal-config
|
|
//#snapshot-config
|
|
akka.persistence.snapshot-store.local.dir = "target/snapshots"
|
|
//#snapshot-config
|
|
"""
|
|
}
|
|
|
|
class PersistencePluginDocSpec extends WordSpec {
|
|
new AnyRef {
|
|
val providerConfig =
|
|
"""
|
|
//#journal-plugin-config
|
|
# Path to the journal plugin to be used
|
|
akka.persistence.journal.plugin = "my-journal"
|
|
|
|
# My custom journal plugin
|
|
my-journal {
|
|
# Class name of the plugin.
|
|
class = "docs.persistence.MyJournal"
|
|
# Dispatcher for the plugin actor.
|
|
plugin-dispatcher = "akka.actor.default-dispatcher"
|
|
}
|
|
//#journal-plugin-config
|
|
|
|
//#snapshot-store-plugin-config
|
|
# Path to the snapshot store plugin to be used
|
|
akka.persistence.snapshot-store.plugin = "my-snapshot-store"
|
|
|
|
# My custom snapshot store plugin
|
|
my-snapshot-store {
|
|
# Class name of the plugin.
|
|
class = "docs.persistence.MySnapshotStore"
|
|
# Dispatcher for the plugin actor.
|
|
plugin-dispatcher = "akka.persistence.dispatchers.default-plugin-dispatcher"
|
|
}
|
|
//#snapshot-store-plugin-config
|
|
"""
|
|
|
|
val system = ActorSystem("doc", ConfigFactory.parseString(providerConfig).withFallback(ConfigFactory.parseString(PersistencePluginDocSpec.config)))
|
|
val extension = Persistence(system)
|
|
}
|
|
}
|
|
|
|
object SharedLeveldbPluginDocSpec {
|
|
import akka.actor._
|
|
import akka.persistence.journal.leveldb.SharedLeveldbJournal
|
|
|
|
val config =
|
|
"""
|
|
//#shared-journal-config
|
|
akka.persistence.journal.plugin = "akka.persistence.journal.leveldb-shared"
|
|
//#shared-journal-config
|
|
//#shared-store-config
|
|
akka.persistence.journal.leveldb-shared.store.dir = "target/shared"
|
|
//#shared-store-config
|
|
"""
|
|
|
|
//#shared-store-usage
|
|
trait SharedStoreUsage extends Actor {
|
|
override def preStart(): Unit = {
|
|
context.actorSelection("akka.tcp://example@127.0.0.1:2552/user/store") ! Identify(1)
|
|
}
|
|
|
|
def receive = {
|
|
case ActorIdentity(1, Some(store)) ⇒
|
|
SharedLeveldbJournal.setStore(store, context.system)
|
|
}
|
|
}
|
|
//#shared-store-usage
|
|
}
|
|
|
|
trait SharedLeveldbPluginDocSpec {
|
|
val system: ActorSystem
|
|
|
|
new AnyRef {
|
|
import akka.actor._
|
|
//#shared-store-creation
|
|
import akka.persistence.journal.leveldb.SharedLeveldbStore
|
|
|
|
val store = system.actorOf(Props[SharedLeveldbStore], "store")
|
|
//#shared-store-creation
|
|
}
|
|
}
|
|
|
|
class MyJournal extends AsyncWriteJournal {
|
|
def writeAsync(persistentBatch: Seq[PersistentRepr]): Future[Unit] = ???
|
|
def deleteAsync(processorId: String, fromSequenceNr: Long, toSequenceNr: Long, permanent: Boolean): Future[Unit] = ???
|
|
def confirmAsync(processorId: String, sequenceNr: Long, channelId: String): Future[Unit] = ???
|
|
def replayAsync(processorId: String, fromSequenceNr: Long, toSequenceNr: Long)(replayCallback: (PersistentRepr) ⇒ Unit): Future[Long] = ???
|
|
}
|
|
|
|
class MySnapshotStore extends SnapshotStore {
|
|
def loadAsync(processorId: String, criteria: SnapshotSelectionCriteria): Future[Option[SelectedSnapshot]] = ???
|
|
def saveAsync(metadata: SnapshotMetadata, snapshot: Any): Future[Unit] = ???
|
|
def saved(metadata: SnapshotMetadata): Unit = ???
|
|
def delete(metadata: SnapshotMetadata): Unit = ???
|
|
def delete(processorId: String, criteria: SnapshotSelectionCriteria): Unit = ???
|
|
}
|