You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pekko.apache.org by md...@apache.org on 2022/11/11 19:03:48 UTC
[incubator-pekko-persistence-dynamodb] branch main updated (d97b1c4 -> 5974c87)
This is an automated email from the ASF dual-hosted git repository.
mdedetrich pushed a change to branch main
in repository https://gitbox.apache.org/repos/asf/incubator-pekko-persistence-dynamodb.git
from d97b1c4 Merge pull request #1 from pjfanning/asf-yaml
new 04418b6 Update scalafmt
new eb7a1cd Add .gitattributes to enforce unix line endings
new 5974c87 format source with scalafmt, #2
The 3 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.gitattributes | 5 ++
.scalafmt.conf | 55 +++++++++++++++-------
build.sbt | 32 ++++++-------
project/Publish.scala | 2 +-
project/plugins.sbt | 10 ++--
.../akka/persistence/dynamodb/DynamoDBConfig.scala | 4 +-
.../persistence/dynamodb/DynamoDBRequests.scala | 2 +-
.../dynamodb/journal/DynamoDBHelper.scala | 4 +-
.../dynamodb/journal/DynamoDBJournal.scala | 20 ++++----
.../dynamodb/journal/DynamoDBJournalConfig.scala | 26 +++++-----
.../dynamodb/journal/DynamoDBJournalRequests.scala | 30 ++++++------
.../dynamodb/journal/DynamoDBRecovery.scala | 52 ++++++++++----------
.../persistence/dynamodb/journal/package.scala | 18 +++----
.../scala/akka/persistence/dynamodb/package.scala | 12 ++---
.../dynamodb/query/DynamoDBReadJournalConfig.scala | 14 +++---
.../query/DynamoDBReadJournalProvider.scala | 4 +-
.../query/javadsl/DynamoDBReadJournal.scala | 2 +-
.../DynamoDBCurrentPersistenceIdsQuery.scala | 6 +--
.../scaladsl/internal/DynamoDBReadJournal.scala | 8 ++--
.../dynamodb/snapshot/DynamoDBSnapshotConfig.scala | 14 +++---
.../snapshot/DynamoDBSnapshotRequests.scala | 19 ++++----
.../dynamodb/snapshot/DynamoDBSnapshotStore.scala | 4 +-
.../persistence/dynamodb/snapshot/package.scala | 12 ++---
.../journal/BackwardsCompatibilityV1Spec.scala | 10 ++--
.../dynamodb/journal/DeletionSpec.scala | 16 +++----
.../journal/DynamoDBIntegrationLoadSpec.scala | 24 +++++-----
.../dynamodb/journal/DynamoDBUtils.scala | 4 +-
.../dynamodb/journal/FailureReportingSpec.scala | 22 +++++----
.../journal/PartialAsyncSerializationSpec.scala | 2 +-
.../journal/PersistAllConsistencySpec.scala | 10 ++--
.../dynamodb/journal/RecoveryConsistencySpec.scala | 20 ++++----
.../dynamodb/journal/WriteThroughputBench.scala | 22 ++++-----
.../dynamodb/query/CurrentPersistenceIdsSpec.scala | 24 +++++-----
.../dynamodb/snapshot/DynamoDBUtils.scala | 2 +-
34 files changed, 269 insertions(+), 242 deletions(-)
create mode 100644 .gitattributes
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pekko.apache.org
For additional commands, e-mail: commits-help@pekko.apache.org
[incubator-pekko-persistence-dynamodb] 01/03: Update scalafmt
Posted by md...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
mdedetrich pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/incubator-pekko-persistence-dynamodb.git
commit 04418b6cd28f270022c71e3355e5c5014f5f2ba4
Author: Matthew de Detrich <ma...@aiven.io>
AuthorDate: Fri Nov 11 18:00:56 2022 +0100
Update scalafmt
---
.scalafmt.conf | 55 +++++++++++++++++++++++++++++++++++++----------------
project/plugins.sbt | 2 +-
2 files changed, 40 insertions(+), 17 deletions(-)
diff --git a/.scalafmt.conf b/.scalafmt.conf
index 0cabdd6..2ca4ece 100644
--- a/.scalafmt.conf
+++ b/.scalafmt.conf
@@ -1,18 +1,40 @@
-version = 2.6.4
-
-style = defaultWithAlign
-
-docstrings = JavaDoc
-indentOperator = spray
-maxColumn = 120
-rewrite.rules = [RedundantParens, SortImports, AvoidInfix]
-unindentTopLevelOperators = true
-align.preset = more
-optIn.breakChainOnFirstMethodDot = false
-optIn.configStyleArguments = false
-danglingParentheses = false
-spaces.inImportCurlyBraces = true
-rewrite.neverInfix.excludeFilters = [
+version = 3.6.1
+runner.dialect = scala213
+project.git = true
+style = defaultWithAlign
+docstrings.style = Asterisk
+docstrings.wrap = false
+indentOperator.preset = spray
+maxColumn = 120
+lineEndings = preserve
+rewrite.rules = [RedundantParens, SortImports, AvoidInfix]
+indentOperator.exemptScope = all
+align.preset = some
+align.tokens."+" = [
+ {
+ code = "~>"
+ owners = [
+ { regex = "Term.ApplyInfix" }
+ ]
+ }
+]
+literals.hexDigits = upper
+literals.hexPrefix = lower
+binPack.unsafeCallSite = always
+binPack.unsafeDefnSite = always
+binPack.indentCallSiteSingleArg = false
+binPack.indentCallSiteOnce = true
+newlines.avoidForSimpleOverflow = [slc]
+newlines.source = keep
+newlines.beforeMultiline = keep
+align.openParenDefnSite = false
+align.openParenCallSite = false
+align.allowOverflow = true
+optIn.breakChainOnFirstMethodDot = false
+optIn.configStyleArguments = false
+danglingParentheses.preset = false
+spaces.inImportCurlyBraces = true
+rewrite.neverInfix.excludeFilters = [
and
min
max
@@ -44,8 +66,9 @@ rewrite.neverInfix.excludeFilters = [
theSameElementsAs
theSameElementsInOrderAs
]
-rewriteTokens = {
+rewriteTokens = {
"⇒": "=>"
"→": "->"
"←": "<-"
}
+project.layout = StandardConvention
diff --git a/project/plugins.sbt b/project/plugins.sbt
index df26c9e..952b085 100644
--- a/project/plugins.sbt
+++ b/project/plugins.sbt
@@ -4,6 +4,6 @@ addSbtPlugin("com.github.gseitz" % "sbt-release" % "1.0.9")
addSbtPlugin("com.jsuereth" % "sbt-pgp" % "2.0.1")
addSbtPlugin("com.lightbend" % "sbt-whitesource" % "0.1.7")
addSbtPlugin("com.typesafe.sbt" % "sbt-git" % "0.9.3")
-addSbtPlugin("org.scalameta" % "sbt-scalafmt" % "2.4.2")
+addSbtPlugin("org.scalameta" % "sbt-scalafmt" % "2.4.6")
// https://github.com/dwijnand/sbt-dynver
addSbtPlugin("com.dwijnand" % "sbt-dynver" % "4.1.1")
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pekko.apache.org
For additional commands, e-mail: commits-help@pekko.apache.org
[incubator-pekko-persistence-dynamodb] 03/03: format source with scalafmt, #2
Posted by md...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
mdedetrich pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/incubator-pekko-persistence-dynamodb.git
commit 5974c87e40c0db5ab3f2109d29db720fa1aeebb8
Author: Auto Format <nobody>
AuthorDate: Fri Nov 11 18:03:45 2022 +0100
format source with scalafmt, #2
---
build.sbt | 32 ++++++-------
project/Publish.scala | 2 +-
project/plugins.sbt | 10 ++---
.../akka/persistence/dynamodb/DynamoDBConfig.scala | 4 +-
.../persistence/dynamodb/DynamoDBRequests.scala | 2 +-
.../dynamodb/journal/DynamoDBHelper.scala | 4 +-
.../dynamodb/journal/DynamoDBJournal.scala | 20 ++++-----
.../dynamodb/journal/DynamoDBJournalConfig.scala | 26 +++++------
.../dynamodb/journal/DynamoDBJournalRequests.scala | 30 ++++++-------
.../dynamodb/journal/DynamoDBRecovery.scala | 52 +++++++++++-----------
.../persistence/dynamodb/journal/package.scala | 18 ++++----
.../scala/akka/persistence/dynamodb/package.scala | 12 ++---
.../dynamodb/query/DynamoDBReadJournalConfig.scala | 14 +++---
.../query/DynamoDBReadJournalProvider.scala | 4 +-
.../query/javadsl/DynamoDBReadJournal.scala | 2 +-
.../DynamoDBCurrentPersistenceIdsQuery.scala | 6 +--
.../scaladsl/internal/DynamoDBReadJournal.scala | 8 ++--
.../dynamodb/snapshot/DynamoDBSnapshotConfig.scala | 14 +++---
.../snapshot/DynamoDBSnapshotRequests.scala | 19 ++++----
.../dynamodb/snapshot/DynamoDBSnapshotStore.scala | 4 +-
.../persistence/dynamodb/snapshot/package.scala | 12 ++---
.../journal/BackwardsCompatibilityV1Spec.scala | 10 ++---
.../dynamodb/journal/DeletionSpec.scala | 16 +++----
.../journal/DynamoDBIntegrationLoadSpec.scala | 24 +++++-----
.../dynamodb/journal/DynamoDBUtils.scala | 4 +-
.../dynamodb/journal/FailureReportingSpec.scala | 22 ++++-----
.../journal/PartialAsyncSerializationSpec.scala | 2 +-
.../journal/PersistAllConsistencySpec.scala | 10 ++---
.../dynamodb/journal/RecoveryConsistencySpec.scala | 20 ++++-----
.../dynamodb/journal/WriteThroughputBench.scala | 22 ++++-----
.../dynamodb/query/CurrentPersistenceIdsSpec.scala | 24 +++++-----
.../dynamodb/snapshot/DynamoDBUtils.scala | 2 +-
32 files changed, 225 insertions(+), 226 deletions(-)
diff --git a/build.sbt b/build.sbt
index 6f72085..37aeb65 100644
--- a/build.sbt
+++ b/build.sbt
@@ -4,25 +4,25 @@ scalaVersion := "2.13.7"
crossScalaVersions := Seq("2.12.13", "2.12.15", "2.13.5", "2.13.7")
crossVersion := CrossVersion.binary
-val akkaVersion = "2.5.29"
-val amzVersion = "1.12.286"
+val akkaVersion = "2.5.29"
+val amzVersion = "1.12.286"
val testcontainersScalaVersion = "0.40.10"
libraryDependencies ++= Seq(
- "com.amazonaws" % "aws-java-sdk-core" % amzVersion,
- "com.amazonaws" % "aws-java-sdk-dynamodb" % amzVersion,
- "javax.xml.bind" % "jaxb-api" % "2.3.1", // see https://github.com/seek-oss/gradle-aws-plugin/issues/15
- "com.typesafe.akka" %% "akka-persistence" % akkaVersion,
- "com.typesafe.akka" %% "akka-persistence-query" % akkaVersion,
- "com.typesafe.akka" %% "akka-stream" % akkaVersion,
- "org.scala-lang.modules" %% "scala-collection-compat" % "2.6.0",
- "com.typesafe.akka" %% "akka-persistence-tck" % akkaVersion % "test",
- "com.typesafe.akka" %% "akka-testkit" % akkaVersion % "test",
- "com.typesafe.akka" %% "akka-stream-testkit" % akkaVersion % "test",
- "org.scalatest" %% "scalatest" % "3.0.8" % "test",
- "commons-io" % "commons-io" % "2.4" % "test",
- "org.hdrhistogram" % "HdrHistogram" % "2.1.8" % "test",
- "com.dimafeng" %% "testcontainers-scala-scalatest" % testcontainersScalaVersion % "test")
+ "com.amazonaws" % "aws-java-sdk-core" % amzVersion,
+ "com.amazonaws" % "aws-java-sdk-dynamodb" % amzVersion,
+ "javax.xml.bind" % "jaxb-api" % "2.3.1", // see https://github.com/seek-oss/gradle-aws-plugin/issues/15
+ "com.typesafe.akka" %% "akka-persistence" % akkaVersion,
+ "com.typesafe.akka" %% "akka-persistence-query" % akkaVersion,
+ "com.typesafe.akka" %% "akka-stream" % akkaVersion,
+ "org.scala-lang.modules" %% "scala-collection-compat" % "2.6.0",
+ "com.typesafe.akka" %% "akka-persistence-tck" % akkaVersion % "test",
+ "com.typesafe.akka" %% "akka-testkit" % akkaVersion % "test",
+ "com.typesafe.akka" %% "akka-stream-testkit" % akkaVersion % "test",
+ "org.scalatest" %% "scalatest" % "3.0.8" % "test",
+ "commons-io" % "commons-io" % "2.4" % "test",
+ "org.hdrhistogram" % "HdrHistogram" % "2.1.8" % "test",
+ "com.dimafeng" %% "testcontainers-scala-scalatest" % testcontainersScalaVersion % "test")
Test / parallelExecution := false
// required by test-containers-scala
diff --git a/project/Publish.scala b/project/Publish.scala
index 9df2352..c8559cc 100644
--- a/project/Publish.scala
+++ b/project/Publish.scala
@@ -12,7 +12,7 @@ object Publish extends AutoPlugin {
val defaultPublishTo = settingKey[File]("Default publish directory")
- override def trigger = allRequirements
+ override def trigger = allRequirements
override def requires = sbtrelease.ReleasePlugin
override lazy val projectSettings = Seq(
diff --git a/project/plugins.sbt b/project/plugins.sbt
index 952b085..2695d22 100644
--- a/project/plugins.sbt
+++ b/project/plugins.sbt
@@ -1,9 +1,9 @@
resolvers += "Typesafe repository".at("https://repo.typesafe.com/typesafe/releases/")
-addSbtPlugin("com.github.gseitz" % "sbt-release" % "1.0.9")
-addSbtPlugin("com.jsuereth" % "sbt-pgp" % "2.0.1")
-addSbtPlugin("com.lightbend" % "sbt-whitesource" % "0.1.7")
-addSbtPlugin("com.typesafe.sbt" % "sbt-git" % "0.9.3")
-addSbtPlugin("org.scalameta" % "sbt-scalafmt" % "2.4.6")
+addSbtPlugin("com.github.gseitz" % "sbt-release" % "1.0.9")
+addSbtPlugin("com.jsuereth" % "sbt-pgp" % "2.0.1")
+addSbtPlugin("com.lightbend" % "sbt-whitesource" % "0.1.7")
+addSbtPlugin("com.typesafe.sbt" % "sbt-git" % "0.9.3")
+addSbtPlugin("org.scalameta" % "sbt-scalafmt" % "2.4.6")
// https://github.com/dwijnand/sbt-dynver
addSbtPlugin("com.dwijnand" % "sbt-dynver" % "4.1.1")
diff --git a/src/main/scala/akka/persistence/dynamodb/DynamoDBConfig.scala b/src/main/scala/akka/persistence/dynamodb/DynamoDBConfig.scala
index 2558b84..46571ad 100644
--- a/src/main/scala/akka/persistence/dynamodb/DynamoDBConfig.scala
+++ b/src/main/scala/akka/persistence/dynamodb/DynamoDBConfig.scala
@@ -38,9 +38,9 @@ class DynamoDBClientConfig(c: Config) extends ClientConfig {
foundSettings ::= s"$path:$value"
}
- private var foundSettings = List.empty[String]
+ private var foundSettings = List.empty[String]
override lazy val toString: String = foundSettings.reverse.mkString("{", ",", "}")
- val config = new ClientConfiguration
+ val config = new ClientConfiguration
get("client-execution-timeout", _.getInt(_), config.setClientExecutionTimeout)
get("connection-max-idle-millis", _.getLong(_), config.setConnectionMaxIdleMillis)
diff --git a/src/main/scala/akka/persistence/dynamodb/DynamoDBRequests.scala b/src/main/scala/akka/persistence/dynamodb/DynamoDBRequests.scala
index ad08bd9..0cab091 100644
--- a/src/main/scala/akka/persistence/dynamodb/DynamoDBRequests.scala
+++ b/src/main/scala/akka/persistence/dynamodb/DynamoDBRequests.scala
@@ -8,7 +8,7 @@ import java.util.{ Collections, HashMap => JHMap, List => JList, Map => JMap }
import akka.Done
import akka.actor.{ Actor, ActorLogging }
-import akka.persistence.dynamodb.journal.{ DynamoDBHelper }
+import akka.persistence.dynamodb.journal.DynamoDBHelper
import com.amazonaws.services.dynamodbv2.model._
import scala.collection.JavaConverters._
import scala.concurrent.Future
diff --git a/src/main/scala/akka/persistence/dynamodb/journal/DynamoDBHelper.scala b/src/main/scala/akka/persistence/dynamodb/journal/DynamoDBHelper.scala
index c1b3d10..9339976 100644
--- a/src/main/scala/akka/persistence/dynamodb/journal/DynamoDBHelper.scala
+++ b/src/main/scala/akka/persistence/dynamodb/journal/DynamoDBHelper.scala
@@ -34,7 +34,7 @@ trait DynamoDBHelper {
def shutdown(): Unit = dynamoDB.shutdown()
- private var reporter: ActorRef = _
+ private var reporter: ActorRef = _
def setReporter(ref: ActorRef): Unit = reporter = ref
private def send[In <: AmazonWebServiceRequest, Out](aws: In, func: AsyncHandler[In, Out] => juc.Future[Out])(implicit
@@ -132,7 +132,7 @@ trait DynamoDBHelper {
def desc(aws: BatchGetItemRequest): String = {
val entry = aws.getRequestItems.entrySet.iterator.next()
val table = entry.getKey
- val keys = entry.getValue.getKeys.asScala.map(formatKey)
+ val keys = entry.getValue.getKeys.asScala.map(formatKey)
s"BatchGetItemRequest($table, ${keys.mkString("(", ",", ")")})"
}
}
diff --git a/src/main/scala/akka/persistence/dynamodb/journal/DynamoDBJournal.scala b/src/main/scala/akka/persistence/dynamodb/journal/DynamoDBJournal.scala
index 8932cd3..d6ff86b 100644
--- a/src/main/scala/akka/persistence/dynamodb/journal/DynamoDBJournal.scala
+++ b/src/main/scala/akka/persistence/dynamodb/journal/DynamoDBJournal.scala
@@ -29,7 +29,7 @@ import akka.actor.TypedActor.context
import scala.concurrent.Promise
import akka.persistence.dynamodb._
-class DynamoDBJournalFailure(message: String, cause: Throwable = null) extends RuntimeException(message, cause)
+class DynamoDBJournalFailure(message: String, cause: Throwable = null) extends RuntimeException(message, cause)
class DynamoDBJournalRejection(message: String, cause: Throwable = null) extends RuntimeException(message, cause)
/**
@@ -85,9 +85,9 @@ class DynamoDBJournal(config: Config)
import context.dispatcher
protected implicit val system: ExtendedActorSystem = context.system.asInstanceOf[ExtendedActorSystem]
- implicit val materializer: Materializer = SystemMaterializer(context.system).materializer
+ implicit val materializer: Materializer = SystemMaterializer(context.system).materializer
- val extension = Persistence(context.system)
+ val extension = Persistence(context.system)
val serialization: Serialization = SerializationExtension(context.system)
val journalSettings = new DynamoDBJournalConfig(config)
@@ -108,7 +108,7 @@ class DynamoDBJournal(config: Config)
private val opQueue: JMap[String, Future[Done]] = new JHMap
override def asyncWriteMessages(messages: immutable.Seq[AtomicWrite]): Future[immutable.Seq[Try[Unit]]] = {
- val p = Promise[Done]
+ val p = Promise[Done]
val pid = messages.head.persistenceId
opQueue.put(pid, p.future)
@@ -146,10 +146,10 @@ class DynamoDBJournal(config: Config)
override def asyncDeleteMessagesTo(persistenceId: String, toSequenceNr: Long): Future[Unit] =
logFailure(s"delete($persistenceId, upto=$toSequenceNr)") {
log.debug("delete-messages persistenceId={} to={}", persistenceId, toSequenceNr)
- val lowF = readSequenceNr(persistenceId, highest = false)
+ val lowF = readSequenceNr(persistenceId, highest = false)
val highF = readSequenceNr(persistenceId, highest = true)
for {
- lowest <- lowF
+ lowest <- lowF
highest <- highF
val upTo = Math.min(toSequenceNr, highest)
_ <- if (upTo + 1 > lowest) setLS(persistenceId, to = upTo + 1) else Future.successful(Done)
@@ -161,7 +161,7 @@ class DynamoDBJournal(config: Config)
private def listAll(persistenceId: String): Future[ListAllResult] =
for {
- low <- readAllSequenceNr(persistenceId, highest = false)
+ low <- readAllSequenceNr(persistenceId, highest = false)
high <- readAllSequenceNr(persistenceId, highest = true)
seqs <- listAllSeqNr(persistenceId)
} yield ListAllResult(persistenceId, low, high, seqs)
@@ -169,9 +169,9 @@ class DynamoDBJournal(config: Config)
private def purge(persistenceId: String): Future[Done] =
for {
highest <- readSequenceNr(persistenceId, highest = true)
- _ <- deleteMessages(persistenceId, 0, highest)
- _ <- removeLS(persistenceId)
- _ <- removeHS(persistenceId)
+ _ <- deleteMessages(persistenceId, 0, highest)
+ _ <- removeLS(persistenceId)
+ _ <- removeHS(persistenceId)
} yield Done
override def receivePluginInternal = {
diff --git a/src/main/scala/akka/persistence/dynamodb/journal/DynamoDBJournalConfig.scala b/src/main/scala/akka/persistence/dynamodb/journal/DynamoDBJournalConfig.scala
index b04d4ea..1b65cb9 100644
--- a/src/main/scala/akka/persistence/dynamodb/journal/DynamoDBJournalConfig.scala
+++ b/src/main/scala/akka/persistence/dynamodb/journal/DynamoDBJournalConfig.scala
@@ -8,22 +8,22 @@ import com.typesafe.config.Config
import akka.persistence.dynamodb.{ DynamoDBClientConfig, DynamoDBConfig }
class DynamoDBJournalConfig(c: Config) extends DynamoDBConfig {
- val JournalTable = c.getString("journal-table")
- val Table = JournalTable
- val JournalName = c.getString("journal-name")
- val AwsKey = c.getString("aws-access-key-id")
- val AwsSecret = c.getString("aws-secret-access-key")
- val Endpoint = c.getString("endpoint")
- val ReplayDispatcher = c.getString("replay-dispatcher")
- val ClientDispatcher = c.getString("client-dispatcher")
- val SequenceShards = c.getInt("sequence-shards")
+ val JournalTable = c.getString("journal-table")
+ val Table = JournalTable
+ val JournalName = c.getString("journal-name")
+ val AwsKey = c.getString("aws-access-key-id")
+ val AwsSecret = c.getString("aws-secret-access-key")
+ val Endpoint = c.getString("endpoint")
+ val ReplayDispatcher = c.getString("replay-dispatcher")
+ val ClientDispatcher = c.getString("client-dispatcher")
+ val SequenceShards = c.getInt("sequence-shards")
val ReplayParallelism = c.getInt("replay-parallelism")
- val Tracing = c.getBoolean("tracing")
- val LogConfig = c.getBoolean("log-config")
+ val Tracing = c.getBoolean("tracing")
+ val LogConfig = c.getBoolean("log-config")
- val MaxBatchGet = c.getInt("aws-api-limits.max-batch-get")
+ val MaxBatchGet = c.getInt("aws-api-limits.max-batch-get")
val MaxBatchWrite = c.getInt("aws-api-limits.max-batch-write")
- val MaxItemSize = c.getInt("aws-api-limits.max-item-size")
+ val MaxItemSize = c.getInt("aws-api-limits.max-item-size")
object Fixes {
val HighDistrust = c.getBoolean("fixes.high-distrust")
diff --git a/src/main/scala/akka/persistence/dynamodb/journal/DynamoDBJournalRequests.scala b/src/main/scala/akka/persistence/dynamodb/journal/DynamoDBJournalRequests.scala
index 68c0aaa..7b04305 100644
--- a/src/main/scala/akka/persistence/dynamodb/journal/DynamoDBJournalRequests.scala
+++ b/src/main/scala/akka/persistence/dynamodb/journal/DynamoDBJournalRequests.scala
@@ -86,17 +86,17 @@ trait DynamoDBJournalRequests extends DynamoDBRequests {
.sequence(atomicWrite.payload.map(repr => toMsgItem(repr)))
.flatMap { items =>
// we created our writes successfully, send them off to DynamoDB
- val low = atomicWrite.lowestSequenceNr
+ val low = atomicWrite.lowestSequenceNr
val high = atomicWrite.highestSequenceNr
- val id = atomicWrite.persistenceId
+ val id = atomicWrite.persistenceId
val size = N(high - low)
val writes = items.iterator.zipWithIndex.map {
- case (item, index) =>
- item.put(AtomIndex, N(index))
- item.put(AtomEnd, size)
- putReq(item)
- } ++ (if ((low - 1) / PartitionSize != high / PartitionSize) Some(putReq(toHSItem(id, high))) else None)
+ case (item, index) =>
+ item.put(AtomIndex, N(index))
+ item.put(AtomEnd, size)
+ putReq(item)
+ } ++ (if ((low - 1) / PartitionSize != high / PartitionSize) Some(putReq(toHSItem(id, high))) else None)
val futures = writes.grouped(MaxBatchWrite).map { batch =>
dynamo.batchWriteItem(batchWriteReq(batch)).flatMap(r => sendUnprocessedItems(r))
@@ -151,7 +151,7 @@ trait DynamoDBJournalRequests extends DynamoDBRequests {
} yield Seq(putReq(item), putReq(toHSItem(repr.persistenceId, repr.sequenceNr)))
}
.map { writesSeq =>
- val writes = writesSeq.flatten
+ val writes = writesSeq.flatten
val reqItems = Collections.singletonMap(JournalTable, writes.asJava)
batchWriteReq(reqItems)
}
@@ -164,7 +164,7 @@ trait DynamoDBJournalRequests extends DynamoDBRequests {
private def toMsgItem(repr: PersistentRepr): Future[Item] = {
try {
val reprPayload: AnyRef = repr.payload.asInstanceOf[AnyRef]
- val serializer = serialization.serializerFor(reprPayload.getClass)
+ val serializer = serialization.serializerFor(reprPayload.getClass)
val fut = serializer match {
case aS: AsyncSerializer =>
Serialization.withTransportInformation(context.system.asInstanceOf[ExtendedActorSystem]) { () =>
@@ -177,18 +177,18 @@ trait DynamoDBJournalRequests extends DynamoDBRequests {
}
fut.map { serialized =>
- val eventData = B(serialized)
+ val eventData = B(serialized)
val serializerId = N(serializer.identifier)
val fieldLength = repr.persistenceId.getBytes.length + repr.sequenceNr.toString.getBytes.length +
repr.writerUuid.getBytes.length + repr.manifest.getBytes.length
- val manifest = Serializers.manifestFor(serializer, reprPayload)
+ val manifest = Serializers.manifestFor(serializer, reprPayload)
val manifestLength = if (manifest.isEmpty) 0 else manifest.getBytes.length
val itemSize = keyLength(
- repr.persistenceId,
- repr.sequenceNr) + eventData.getB.remaining + serializerId.getN.getBytes.length + manifestLength + fieldLength
+ repr.persistenceId,
+ repr.sequenceNr) + eventData.getB.remaining + serializerId.getN.getBytes.length + manifestLength + fieldLength
if (itemSize > MaxItemSize) {
throw new DynamoDBJournalRejection(s"MaxItemSize exceeded: $itemSize > $MaxItemSize")
@@ -220,7 +220,7 @@ trait DynamoDBJournalRequests extends DynamoDBRequests {
* see the implementation of readSequenceNr for details.
*/
private def toHSItem(persistenceId: String, sequenceNr: Long): Item = {
- val seq = sequenceNr / PartitionSize
+ val seq = sequenceNr / PartitionSize
val item: Item = highSeqKey(persistenceId, seq % SequenceShards)
item.put(SequenceNr, N(seq * PartitionSize))
item
@@ -239,7 +239,7 @@ trait DynamoDBJournalRequests extends DynamoDBRequests {
* because replay must start exactly here.
*/
private def toLSItem(persistenceId: String, sequenceNr: Long): Item = {
- val seq = sequenceNr / PartitionSize
+ val seq = sequenceNr / PartitionSize
val item: Item = lowSeqKey(persistenceId, seq % SequenceShards)
item.put(SequenceNr, N(sequenceNr))
item
diff --git a/src/main/scala/akka/persistence/dynamodb/journal/DynamoDBRecovery.scala b/src/main/scala/akka/persistence/dynamodb/journal/DynamoDBRecovery.scala
index 94e8e71..193dfe9 100644
--- a/src/main/scala/akka/persistence/dynamodb/journal/DynamoDBRecovery.scala
+++ b/src/main/scala/akka/persistence/dynamodb/journal/DynamoDBRecovery.scala
@@ -38,7 +38,7 @@ object DynamoDBRecovery {
case class ReplayBatch(items: Seq[Item], map: Map[AttributeValue, Long]) {
def sorted: immutable.Iterable[Item] =
items.foldLeft(immutable.TreeMap.empty[Long, Item])((acc, i) => acc.updated(itemToSeq(i), i)).map(_._2)
- def ids: Seq[Long] = items.map(itemToSeq).sorted
+ def ids: Seq[Long] = items.map(itemToSeq).sorted
private def itemToSeq(i: Item): Long = map(i.get(Key)) * PartitionSize + i.get(Sort).getN.toInt
}
}
@@ -57,7 +57,7 @@ case class PartitionKeys(partitionSeqNum: Long, partitionEventNums: immutable.Se
*/
object DynamoPartitionGrouped extends GraphStage[FlowShape[Long, PartitionKeys]] {
- val in = Inlet[Long]("DynamoEventNum.in")
+ val in = Inlet[Long]("DynamoEventNum.in")
val out = Outlet[PartitionKeys]("DynamoPartitionKeys.out")
override val initialAttributes = Attributes.name("DynamoPartitionGrouped")
@@ -84,7 +84,7 @@ object DynamoPartitionGrouped extends GraphStage[FlowShape[Long, PartitionKeys]]
partitionBuf += currentSeqNo
hasElements = true
- //If the next entry received would result in the next partition, then we clear the buf and push the results out
+ // If the next entry received would result in the next partition, then we clear the buf and push the results out
if ((currentSeqNo + 1) % PartitionSize == 0) {
val partitionGroup = partitionBuf.result()
pushOut(currentSeqNo, partitionGroup)
@@ -96,11 +96,11 @@ object DynamoPartitionGrouped extends GraphStage[FlowShape[Long, PartitionKeys]]
override def onPull(): Unit = pull(in)
override def onUpstreamFinish(): Unit = {
- //this means the partitionBuf has elements but not a full amount (n). However, since upstream is finished
- //publishing elements, we need to push what we have downstream.
+ // this means the partitionBuf has elements but not a full amount (n). However, since upstream is finished
+ // publishing elements, we need to push what we have downstream.
if (hasElements) {
val partitionGroup = partitionBuf.result()
- val currentSeqNo = partitionGroup.last
+ val currentSeqNo = partitionGroup.last
pushOut(currentSeqNo, partitionGroup)
}
completeStage()
@@ -114,16 +114,16 @@ object DynamoPartitionGrouped extends GraphStage[FlowShape[Long, PartitionKeys]]
object RemoveIncompleteAtoms extends GraphStage[FlowShape[Item, List[Item]]] {
private final val NoBatch = -1L
- val in = Inlet[Item]("RIA.in")
+ val in = Inlet[Item]("RIA.in")
val out = Outlet[List[Item]]("RIA.out")
- override val shape = FlowShape(in, out)
+ override val shape = FlowShape(in, out)
override val initialAttributes = Attributes.name("RemoveIncompleteAtoms")
override def createLogic(attr: Attributes) =
new GraphStageLogic(shape) with InHandler with OutHandler {
var batchEnd = NoBatch
- var batch = List.empty[Item]
+ var batch = List.empty[Item]
setHandler(out, this)
setHandler(in, this)
@@ -133,9 +133,9 @@ object RemoveIncompleteAtoms extends GraphStage[FlowShape[Item, List[Item]]] {
override def onPush(): Unit = {
val item = grab(in)
if (item.containsKey(AtomEnd)) {
- val end = item.get(AtomEnd).getN.toLong
- val index = item.get(AtomIndex).getN.toLong
- val seqNr = sn(item)
+ val end = item.get(AtomEnd).getN.toLong
+ val index = item.get(AtomIndex).getN.toLong
+ val seqNr = sn(item)
val myBatchEnd = seqNr - index + end
if (seqNr == batchEnd) {
val result =
@@ -173,8 +173,8 @@ object RemoveIncompleteAtoms extends GraphStage[FlowShape[Item, List[Item]]] {
}
private def sn(item: Item): Long = {
- val s = item.get(Key).getS
- val n = item.get(Sort).getN.toLong
+ val s = item.get(Key).getS
+ val n = item.get(Sort).getN.toLong
val pos = s.lastIndexOf('-')
require(pos != -1, "unknown key format " + s)
s.substring(pos + 1).toLong * PartitionSize + n
@@ -237,9 +237,9 @@ trait DynamoDBRecovery extends AsyncReplayMessages {
}
def getPartitionItems(persistenceId: String, partitionKeys: PartitionKeys): Future[ReplayBatch] = {
- val sortedNrs = partitionKeys.partitionEventNums.sorted.map(_ % PartitionSize)
+ val sortedNrs = partitionKeys.partitionEventNums.sorted.map(_ % PartitionSize)
val startSortKey = sortedNrs.head
- val endSortKey = sortedNrs.last
+ val endSortKey = sortedNrs.last
val queryRequestBuilder: (Option[java.util.Map[String, AttributeValue]]) => QueryRequest = exclusiveStartKeyOpt => {
val request = new QueryRequest()
@@ -247,9 +247,9 @@ trait DynamoDBRecovery extends AsyncReplayMessages {
.withKeyConditionExpression(s"$Key = :kkey AND $Sort BETWEEN :startSKey AND :endSKey")
.withExpressionAttributeValues(
Map(
- ":kkey" -> S(messagePartitionKeyFromGroupNr(persistenceId, partitionKeys.partitionSeqNum)),
+ ":kkey" -> S(messagePartitionKeyFromGroupNr(persistenceId, partitionKeys.partitionSeqNum)),
":startSKey" -> N(startSortKey),
- ":endSKey" -> N(endSortKey)).asJava)
+ ":endSKey" -> N(endSortKey)).asJava)
.withProjectionExpression(ItemAttributesForReplay.mkString(","))
.withConsistentRead(true)
.withReturnConsumedCapacity(ReturnConsumedCapacity.TOTAL)
@@ -267,7 +267,7 @@ trait DynamoDBRecovery extends AsyncReplayMessages {
}
}
- val batchKeys = partitionKeys.partitionEventNums.map(s => messageKey(persistenceId, s) -> (s / PartitionSize))
+ val batchKeys = partitionKeys.partitionEventNums.map(s => messageKey(persistenceId, s) -> (s / PartitionSize))
val batchKeysMap = batchKeys.iterator.map(p => p._1.get(Key) -> p._2).toMap
dynamoSummingPager(queryRequestBuilder(None), Seq.empty).map(result => ReplayBatch(result, batchKeysMap))
}
@@ -322,7 +322,7 @@ trait DynamoDBRecovery extends AsyncReplayMessages {
* `partitionStart` is the Sort=0 entry’s sequence number, so add the maximum sort key.
*/
val partitionMax = nextResults.getItems.asScala.map(_.get(Sort).getN.toLong).max
- val ret = partitionStart + partitionMax
+ val ret = partitionStart + partitionMax
if (partitionMax == PartitionSize - 1) {
val nextStart = ret + 1
@@ -374,7 +374,7 @@ trait DynamoDBRecovery extends AsyncReplayMessages {
.map(l => if (highest) highSeqKey(persistenceId, l) else lowSeqKey(persistenceId, l))
.grouped(MaxBatchGet)
.map { keys =>
- val ka = new KeysAndAttributes().withKeys(keys.asJava).withConsistentRead(true)
+ val ka = new KeysAndAttributes().withKeys(keys.asJava).withConsistentRead(true)
val get = batchGetReq(Collections.singletonMap(JournalTable, ka))
dynamo.batchGetItem(get).flatMap(getUnprocessedItems(_))
}
@@ -424,13 +424,13 @@ trait DynamoDBRecovery extends AsyncReplayMessages {
if (item.containsKey(Event)) {
val serializerManifest = getValueOrEmptyString(item, SerializerManifest)
- val pI = item.get(PersistentId).getS
- val sN = item.get(SequenceNr).getN.toLong
- val wU = item.get(WriterUuid).getS
+ val pI = item.get(PersistentId).getS
+ val sN = item.get(SequenceNr).getN.toLong
+ val wU = item.get(WriterUuid).getS
val reprManifest = getValueOrEmptyString(item, Manifest)
val eventPayload = item.get(Event).getB
- val serId = item.get(SerializerId).getN.toInt
+ val serId = item.get(SerializerId).getN.toInt
val fut = serialization.serializerByIdentity.get(serId) match {
case Some(asyncSerializer: AsyncSerializer) =>
@@ -486,7 +486,7 @@ trait DynamoDBRecovery extends AsyncReplayMessages {
dynamo
.batchGetItem(rest)
.map { rr =>
- val items = rr.getResponses.get(JournalTable)
+ val items = rr.getResponses.get(JournalTable)
val responses = result.getResponses.get(JournalTable)
items.forEach(new Consumer[Item] {
override def accept(item: Item): Unit = responses.add(item)
diff --git a/src/main/scala/akka/persistence/dynamodb/journal/package.scala b/src/main/scala/akka/persistence/dynamodb/journal/package.scala
index 83d439d..63c9c4c 100644
--- a/src/main/scala/akka/persistence/dynamodb/journal/package.scala
+++ b/src/main/scala/akka/persistence/dynamodb/journal/package.scala
@@ -8,28 +8,28 @@ import com.amazonaws.services.dynamodbv2.model._
package object journal {
// field names
- val Key = "par"
- val Sort = "num"
- val Payload = "pay"
+ val Key = "par"
+ val Sort = "num"
+ val Payload = "pay"
val SequenceNr = "seq"
- val AtomIndex = "idx"
- val AtomEnd = "cnt"
+ val AtomIndex = "idx"
+ val AtomEnd = "cnt"
/* PersistenceRepr fields
sequence_nr and persistence_id extracted from the key
*/
val PersistentId = "persistence_id"
- val WriterUuid = "writer_uuid"
+ val WriterUuid = "writer_uuid"
val Manifest = "manifest"
- val Event = "event"
- val SerializerId = "ev_ser_id"
+ val Event = "event"
+ val SerializerId = "ev_ser_id"
val SerializerManifest = "ev_ser_manifest"
val KeyPayloadOverhead = 26 // including fixed parts of partition key and 36 bytes fudge factor
- //This is the size of each partition used on DynamoDB. This value should never change as it will break backwards compatability.
+ // This is the size of each partition used on DynamoDB. This value should never change as it will break backwards compatability.
val PartitionSize: Int = 100
val schema = new CreateTableRequest()
diff --git a/src/main/scala/akka/persistence/dynamodb/package.scala b/src/main/scala/akka/persistence/dynamodb/package.scala
index 42ea2ba..e8fb3a2 100644
--- a/src/main/scala/akka/persistence/dynamodb/package.scala
+++ b/src/main/scala/akka/persistence/dynamodb/package.scala
@@ -18,14 +18,14 @@ import scala.concurrent.{ ExecutionContext, Future, Promise }
import scala.util.{ Failure, Success, Try }
package object dynamodb {
- type Item = JMap[String, AttributeValue]
+ type Item = JMap[String, AttributeValue]
type ItemUpdates = JMap[String, AttributeValueUpdate]
def S(value: String): AttributeValue = new AttributeValue().withS(value)
- def N(value: Long): AttributeValue = new AttributeValue().withN(value.toString)
+ def N(value: Long): AttributeValue = new AttributeValue().withN(value.toString)
def N(value: String): AttributeValue = new AttributeValue().withN(value)
- val Naught = N(0)
+ val Naught = N(0)
def B(value: Array[Byte]): AttributeValue = new AttributeValue().withB(ByteBuffer.wrap(value))
@@ -49,15 +49,15 @@ package object dynamodb {
executor: ExecutionContext): Future[M[Try[A]]] =
in.foldLeft(Future.successful(cbf(in))) { (fr, a) =>
val fb = lift(a)
- for (r <- fr; b <- fb) yield (r += b)
+ for (r <- fr; b <- fb) yield r += b
}.map(_.result())
def dynamoClient(system: ActorSystem, settings: DynamoDBConfig): DynamoDBHelper = {
val client =
if (settings.AwsKey.nonEmpty && settings.AwsSecret.nonEmpty) {
- val conns = settings.client.config.getMaxConnections
+ val conns = settings.client.config.getMaxConnections
val executor = Executors.newFixedThreadPool(conns)
- val creds = new BasicAWSCredentials(settings.AwsKey, settings.AwsSecret)
+ val creds = new BasicAWSCredentials(settings.AwsKey, settings.AwsSecret)
new AmazonDynamoDBAsyncClient(creds, settings.client.config, executor)
} else {
new AmazonDynamoDBAsyncClient(settings.client.config)
diff --git a/src/main/scala/akka/persistence/dynamodb/query/DynamoDBReadJournalConfig.scala b/src/main/scala/akka/persistence/dynamodb/query/DynamoDBReadJournalConfig.scala
index caedb19..5d35a48 100644
--- a/src/main/scala/akka/persistence/dynamodb/query/DynamoDBReadJournalConfig.scala
+++ b/src/main/scala/akka/persistence/dynamodb/query/DynamoDBReadJournalConfig.scala
@@ -5,15 +5,15 @@ import akka.persistence.dynamodb.{ ClientConfig, DynamoDBClientConfig, DynamoDBC
import com.typesafe.config.Config
class DynamoDBReadJournalConfig(c: Config) extends DynamoDBConfig {
- val Table: String = c.getString("journal-table")
+ val Table: String = c.getString("journal-table")
val JournalName: String = c.getString("journal-name")
- val AwsKey: String = c.getString("aws-access-key-id")
- val AwsSecret: String = c.getString("aws-secret-access-key")
- val Endpoint: String = c.getString("endpoint")
+ val AwsKey: String = c.getString("aws-access-key-id")
+ val AwsSecret: String = c.getString("aws-secret-access-key")
+ val Endpoint: String = c.getString("endpoint")
- val MaxBatchGet: Int = c.getInt("aws-api-limits.max-batch-get")
+ val MaxBatchGet: Int = c.getInt("aws-api-limits.max-batch-get")
val MaxBatchWrite: Int = c.getInt("aws-api-limits.max-batch-write")
- val MaxItemSize: Int = c.getInt("aws-api-limits.max-item-size")
+ val MaxItemSize: Int = c.getInt("aws-api-limits.max-item-size")
val PersistenceIdsIndexName: String = c.getString("persistence-ids-index-name")
@@ -26,7 +26,7 @@ class DynamoDBReadJournalConfig(c: Config) extends DynamoDBConfig {
override val client: ClientConfig = new DynamoDBClientConfig(c)
override val ClientDispatcher: String = c.getString("client-dispatcher")
- override val Tracing: Boolean = c.getBoolean("tracing")
+ override val Tracing: Boolean = c.getBoolean("tracing")
}
object DynamoDBReadJournalConfig {
diff --git a/src/main/scala/akka/persistence/dynamodb/query/DynamoDBReadJournalProvider.scala b/src/main/scala/akka/persistence/dynamodb/query/DynamoDBReadJournalProvider.scala
index 95097b4..8dfaa7b 100644
--- a/src/main/scala/akka/persistence/dynamodb/query/DynamoDBReadJournalProvider.scala
+++ b/src/main/scala/akka/persistence/dynamodb/query/DynamoDBReadJournalProvider.scala
@@ -10,10 +10,10 @@ import com.typesafe.config.Config
class DynamoDBReadJournalProvider(system: ExtendedActorSystem, config: Config, configPath: String)
extends ReadJournalProvider {
- private lazy val _scalaReadJournal = new ScalaDynamoDBReadJournal(config, configPath)(system)
+ private lazy val _scalaReadJournal = new ScalaDynamoDBReadJournal(config, configPath)(system)
override def scaladslReadJournal(): ScalaReadJournal = _scalaReadJournal
- private lazy val _javadslReadJournal = new JavaDynamoDBReadJournal(_scalaReadJournal)
+ private lazy val _javadslReadJournal = new JavaDynamoDBReadJournal(_scalaReadJournal)
override def javadslReadJournal(): JavaReadJournal = _javadslReadJournal
}
diff --git a/src/main/scala/akka/persistence/dynamodb/query/javadsl/DynamoDBReadJournal.scala b/src/main/scala/akka/persistence/dynamodb/query/javadsl/DynamoDBReadJournal.scala
index 21014f4..2efe215 100644
--- a/src/main/scala/akka/persistence/dynamodb/query/javadsl/DynamoDBReadJournal.scala
+++ b/src/main/scala/akka/persistence/dynamodb/query/javadsl/DynamoDBReadJournal.scala
@@ -38,7 +38,7 @@ class DynamoDBReadJournal(scaladslReadJournal: scaladsl.DynamoDBReadJournal)
// with akka.persistence.query.javadsl.EventsByTagQuery
with CurrentEventsByPersistenceIdQuery
with CurrentPersistenceIdsQuery
- //with akka.persistence.query.javadsl.CurrentPersistenceIdsQuery
+ // with akka.persistence.query.javadsl.CurrentPersistenceIdsQuery
{
/**
diff --git a/src/main/scala/akka/persistence/dynamodb/query/scaladsl/internal/DynamoDBCurrentPersistenceIdsQuery.scala b/src/main/scala/akka/persistence/dynamodb/query/scaladsl/internal/DynamoDBCurrentPersistenceIdsQuery.scala
index 66ed86a..09eb1a6 100644
--- a/src/main/scala/akka/persistence/dynamodb/query/scaladsl/internal/DynamoDBCurrentPersistenceIdsQuery.scala
+++ b/src/main/scala/akka/persistence/dynamodb/query/scaladsl/internal/DynamoDBCurrentPersistenceIdsQuery.scala
@@ -91,7 +91,7 @@ trait DynamoDBCurrentPersistenceIdsQuery extends PublicDynamoDBCurrentPersistenc
def nextCall(maybePreviousResult: Option[Result]): Future[Option[Result]] = {
val maybeNextResult = for {
- previousResult <- maybePreviousResult
+ previousResult <- maybePreviousResult
nextEvaluatedKey <- previousResult.nextEvaluatedKey
} yield getPersistenceIds(Some(nextEvaluatedKey)).map(Some(_))
@@ -159,8 +159,8 @@ trait DynamoDBCurrentPersistenceIdsQuery extends PublicDynamoDBCurrentPersistenc
try {
val prefixLength = journalName.length + 3
val startPostfix = rawPersistenceId.lastIndexOf("-")
- val postfix = rawPersistenceId.substring(startPostfix)
- val partitionNr = postfix.substring(postfix.lastIndexOf("-")).toInt
+ val postfix = rawPersistenceId.substring(startPostfix)
+ val partitionNr = postfix.substring(postfix.lastIndexOf("-")).toInt
if (partitionNr == 0)
Some(rawPersistenceId.substring(prefixLength, startPostfix))
else
diff --git a/src/main/scala/akka/persistence/dynamodb/query/scaladsl/internal/DynamoDBReadJournal.scala b/src/main/scala/akka/persistence/dynamodb/query/scaladsl/internal/DynamoDBReadJournal.scala
index f6de391..8971c65 100644
--- a/src/main/scala/akka/persistence/dynamodb/query/scaladsl/internal/DynamoDBReadJournal.scala
+++ b/src/main/scala/akka/persistence/dynamodb/query/scaladsl/internal/DynamoDBReadJournal.scala
@@ -29,11 +29,11 @@ class DynamoDBReadJournal(config: Config, configPath: String)(implicit val syste
with SerializationProvider
with ActorSystemLoggingProvider {
- protected val readJournalSettings = new DynamoDBReadJournalConfig(config)
- protected val dynamo: DynamoDBHelper = dynamoClient(system, readJournalSettings)
- val serialization: Serialization = SerializationExtension(system)
+ protected val readJournalSettings = new DynamoDBReadJournalConfig(config)
+ protected val dynamo: DynamoDBHelper = dynamoClient(system, readJournalSettings)
+ val serialization: Serialization = SerializationExtension(system)
implicit val materializer: Materializer = SystemMaterializer(system).materializer
- val journalSettings = new DynamoDBJournalConfig(config)
+ val journalSettings = new DynamoDBJournalConfig(config)
def close(): Unit = dynamo.shutdown()
}
diff --git a/src/main/scala/akka/persistence/dynamodb/snapshot/DynamoDBSnapshotConfig.scala b/src/main/scala/akka/persistence/dynamodb/snapshot/DynamoDBSnapshotConfig.scala
index f54c102..ac90075 100644
--- a/src/main/scala/akka/persistence/dynamodb/snapshot/DynamoDBSnapshotConfig.scala
+++ b/src/main/scala/akka/persistence/dynamodb/snapshot/DynamoDBSnapshotConfig.scala
@@ -7,15 +7,15 @@ import akka.persistence.dynamodb.{ ClientConfig, DynamoDBClientConfig, DynamoDBC
import com.typesafe.config.Config
class DynamoDBSnapshotConfig(c: Config) extends DynamoDBConfig {
- val Table = c.getString("snapshot-table")
+ val Table = c.getString("snapshot-table")
val JournalName = c.getString("journal-name")
- val AwsKey = c.getString("aws-access-key-id")
- val AwsSecret = c.getString("aws-secret-access-key")
- val Endpoint = c.getString("endpoint")
+ val AwsKey = c.getString("aws-access-key-id")
+ val AwsSecret = c.getString("aws-secret-access-key")
+ val Endpoint = c.getString("endpoint")
- val MaxBatchGet = c.getInt("aws-api-limits.max-batch-get")
+ val MaxBatchGet = c.getInt("aws-api-limits.max-batch-get")
val MaxBatchWrite = c.getInt("aws-api-limits.max-batch-write")
- val MaxItemSize = c.getInt("aws-api-limits.max-item-size")
+ val MaxItemSize = c.getInt("aws-api-limits.max-item-size")
override def toString: String =
"DynamoDBJournalConfig(" +
@@ -26,6 +26,6 @@ class DynamoDBSnapshotConfig(c: Config) extends DynamoDBConfig {
override val client: ClientConfig = new DynamoDBClientConfig(c)
override val ClientDispatcher = c.getString("client-dispatcher")
- override val Tracing = c.getBoolean("tracing")
+ override val Tracing = c.getBoolean("tracing")
}
diff --git a/src/main/scala/akka/persistence/dynamodb/snapshot/DynamoDBSnapshotRequests.scala b/src/main/scala/akka/persistence/dynamodb/snapshot/DynamoDBSnapshotRequests.scala
index 89a7896..bf24da5 100644
--- a/src/main/scala/akka/persistence/dynamodb/snapshot/DynamoDBSnapshotRequests.scala
+++ b/src/main/scala/akka/persistence/dynamodb/snapshot/DynamoDBSnapshotRequests.scala
@@ -146,8 +146,8 @@ trait DynamoDBSnapshotRequests extends DynamoDBRequests {
item.put(SequenceNr, N(sequenceNr))
item.put(Timestamp, N(timestamp))
val snapshotData = snapshot.asInstanceOf[AnyRef]
- val serializer = serialization.findSerializerFor(snapshotData)
- val manifest = Serializers.manifestFor(serializer, snapshotData)
+ val serializer = serialization.findSerializerFor(snapshotData)
+ val manifest = Serializers.manifestFor(serializer, snapshotData)
val fut = serializer match {
case asyncSer: AsyncSerializer =>
@@ -172,14 +172,14 @@ trait DynamoDBSnapshotRequests extends DynamoDBRequests {
}
private def fromSnapshotItem(persistenceId: String, item: Item): Future[SelectedSnapshot] = {
- val seqNr = item.get(SequenceNr).getN.toLong
+ val seqNr = item.get(SequenceNr).getN.toLong
val timestamp = item.get(Timestamp).getN.toLong
if (item.containsKey(PayloadData)) {
val payloadData = item.get(PayloadData).getB
- val serId = item.get(SerializerId).getN.toInt
- val manifest = if (item.containsKey(SerializerManifest)) item.get(SerializerManifest).getS else ""
+ val serId = item.get(SerializerId).getN.toInt
+ val manifest = if (item.containsKey(SerializerManifest)) item.get(SerializerManifest).getS else ""
val serialized = serialization.serializerByIdentity(serId) match {
case aS: AsyncSerializer =>
@@ -190,11 +190,10 @@ trait DynamoDBSnapshotRequests extends DynamoDBRequests {
Future.successful(serialization.deserialize(payloadData.array(), serId, manifest).get)
}
- serialized.map(
- data =>
- SelectedSnapshot(
- metadata = SnapshotMetadata(persistenceId, sequenceNr = seqNr, timestamp = timestamp),
- snapshot = data))
+ serialized.map(data =>
+ SelectedSnapshot(
+ metadata = SnapshotMetadata(persistenceId, sequenceNr = seqNr, timestamp = timestamp),
+ snapshot = data))
} else {
val payloadValue = item.get(Payload).getB
diff --git a/src/main/scala/akka/persistence/dynamodb/snapshot/DynamoDBSnapshotStore.scala b/src/main/scala/akka/persistence/dynamodb/snapshot/DynamoDBSnapshotStore.scala
index 00b363f..a752e1f 100644
--- a/src/main/scala/akka/persistence/dynamodb/snapshot/DynamoDBSnapshotStore.scala
+++ b/src/main/scala/akka/persistence/dynamodb/snapshot/DynamoDBSnapshotStore.scala
@@ -14,8 +14,8 @@ import scala.concurrent.Future
class DynamoDBSnapshotStore(config: Config) extends SnapshotStore with DynamoDBSnapshotRequests with ActorLogging {
val journalSettings = new DynamoDBSnapshotConfig(config)
- val dynamo = dynamoClient(context.system, journalSettings)
- val serialization = SerializationExtension(context.system)
+ val dynamo = dynamoClient(context.system, journalSettings)
+ val serialization = SerializationExtension(context.system)
import journalSettings._
override def preStart(): Unit = {
diff --git a/src/main/scala/akka/persistence/dynamodb/snapshot/package.scala b/src/main/scala/akka/persistence/dynamodb/snapshot/package.scala
index e13ff07..411dfec 100644
--- a/src/main/scala/akka/persistence/dynamodb/snapshot/package.scala
+++ b/src/main/scala/akka/persistence/dynamodb/snapshot/package.scala
@@ -6,13 +6,13 @@ package akka.persistence.dynamodb
package object snapshot {
// field names
- val Key = "par"
- val Timestamp = "ts"
+ val Key = "par"
+ val Timestamp = "ts"
val TimestampIndex = "ts-idx"
- val Payload = "pay"
- val SequenceNr = "seq"
- val SerializerId = "ser_id"
+ val Payload = "pay"
+ val SequenceNr = "seq"
+ val SerializerId = "ser_id"
val SerializerManifest = "ser_manifest"
- val PayloadData = "pay_data"
+ val PayloadData = "pay_data"
}
diff --git a/src/test/scala/akka/persistence/dynamodb/journal/BackwardsCompatibilityV1Spec.scala b/src/test/scala/akka/persistence/dynamodb/journal/BackwardsCompatibilityV1Spec.scala
index 00411dc..26ec5dc 100644
--- a/src/test/scala/akka/persistence/dynamodb/journal/BackwardsCompatibilityV1Spec.scala
+++ b/src/test/scala/akka/persistence/dynamodb/journal/BackwardsCompatibilityV1Spec.scala
@@ -31,10 +31,10 @@ class BackwardsCompatibilityV1Spec
with IntegSpec {
def loadV1VersionData(): Unit = {
- val config = ConfigFactory.load()
- val endpoint = config.getString("my-dynamodb-journal.endpoint")
+ val config = ConfigFactory.load()
+ val endpoint = config.getString("my-dynamodb-journal.endpoint")
val tableName = config.getString("my-dynamodb-journal.journal-table")
- val accesKey = config.getString("my-dynamodb-journal.aws-access-key-id")
+ val accesKey = config.getString("my-dynamodb-journal.aws-access-key-id")
val secretKey = config.getString("my-dynamodb-journal.aws-secret-access-key")
val client: AmazonDynamoDB =
@@ -97,12 +97,12 @@ class BackwardsCompatibilityV1Spec
}
override val persistenceId = "OldFormatEvents"
- lazy val journal = Persistence(system).journalFor("")
+ lazy val journal = Persistence(system).journalFor("")
"DynamoDB Journal (Backwards Compatibility Test)" must {
val messages = 20
- val probe = TestProbe()
+ val probe = TestProbe()
s"successfully replay events in old format - created by old version of the plugin" in {
diff --git a/src/test/scala/akka/persistence/dynamodb/journal/DeletionSpec.scala b/src/test/scala/akka/persistence/dynamodb/journal/DeletionSpec.scala
index b3f38b8..14cd785 100644
--- a/src/test/scala/akka/persistence/dynamodb/journal/DeletionSpec.scala
+++ b/src/test/scala/akka/persistence/dynamodb/journal/DeletionSpec.scala
@@ -45,12 +45,12 @@ class DeletionSpec
super.afterAll()
}
- override val persistenceId = "DeletionSpec"
+ override val persistenceId = "DeletionSpec"
implicit val materializer: Materializer = SystemMaterializer(system).materializer
- lazy val journal = Persistence(system).journalFor("")
- lazy val queries = PersistenceQuery(system).readJournalFor[DynamoDBReadJournal](DynamoDBReadJournal.Identifier)
- val msgs = (1 to 149).map(i => s"a-$i")
- val more = (150 to 200).map(i => s"b-$i")
+ lazy val journal = Persistence(system).journalFor("")
+ lazy val queries = PersistenceQuery(system).readJournalFor[DynamoDBReadJournal](DynamoDBReadJournal.Identifier)
+ val msgs = (1 to 149).map(i => s"a-$i")
+ val more = (150 to 200).map(i => s"b-$i")
"DynamoDB Journal (Deletion)" must {
@@ -66,13 +66,13 @@ class DeletionSpec
expectMsg(WriteMessagesSuccessful)
(1 to 149).foreach(i => expectMsgType[WriteMessageSuccess].persistent.sequenceNr.toInt should ===(i))
journal ! ListAll(persistenceId, testActor)
- expectMsg(ListAllResult(persistenceId, Set.empty, Set(100L), (1L to 149)))
+ expectMsg(ListAllResult(persistenceId, Set.empty, Set(100L), 1L to 149))
journal ! WriteMessages(AtomicWrite(more.map(m => persistentRepr(m))) :: Nil, testActor, 1)
expectMsg(WriteMessagesSuccessful)
(150 to 200).foreach(i => expectMsgType[WriteMessageSuccess].persistent.sequenceNr.toInt should ===(i))
journal ! ListAll(persistenceId, testActor)
- expectMsg(ListAllResult(persistenceId, Set.empty, Set(100L, 200L), (1L to 200)))
+ expectMsg(ListAllResult(persistenceId, Set.empty, Set(100L, 200L), 1L to 200))
}
"3 delete some events" in {
@@ -93,7 +93,7 @@ class DeletionSpec
journal ! DeleteMessagesTo(persistenceId, 3L, testActor)
expectMsg(DeleteMessagesSuccess(3L))
journal ! ListAll(persistenceId, testActor)
- expectMsg(ListAllResult(persistenceId, Set(6L), Set(100L, 200L), (6L to 200)))
+ expectMsg(ListAllResult(persistenceId, Set(6L), Set(100L, 200L), 6L to 200))
}
"5 delete all events" in {
diff --git a/src/test/scala/akka/persistence/dynamodb/journal/DynamoDBIntegrationLoadSpec.scala b/src/test/scala/akka/persistence/dynamodb/journal/DynamoDBIntegrationLoadSpec.scala
index 6555c7d..5e2e966 100644
--- a/src/test/scala/akka/persistence/dynamodb/journal/DynamoDBIntegrationLoadSpec.scala
+++ b/src/test/scala/akka/persistence/dynamodb/journal/DynamoDBIntegrationLoadSpec.scala
@@ -173,7 +173,7 @@ class DynamoDBIntegrationLoadSpec
"write and replay messages" in {
val persistenceId = UUID.randomUUID().toString
- val processor1 = system.actorOf(Props(classOf[ProcessorA], persistenceId, self), "p1")
+ val processor1 = system.actorOf(Props(classOf[ProcessorA], persistenceId, self), "p1")
(1L to 16L).foreach { i =>
processor1 ! s"a-${i}"
expectMsgAllOf(s"a-${i}", i, false)
@@ -218,8 +218,8 @@ class DynamoDBIntegrationLoadSpec
probe.expectNoMsg(200.millis)
} */
"write and replay with persistAll greater than partition size skipping whole partition" in {
- val persistenceId = UUID.randomUUID().toString
- val probe = TestProbe()
+ val persistenceId = UUID.randomUUID().toString
+ val probe = TestProbe()
val processorAtomic = system.actorOf(Props(classOf[ProcessorAtomic], persistenceId, self))
processorAtomic ! List("a-1", "a-2", "a-3", "a-4", "a-5", "a-6")
@@ -227,15 +227,15 @@ class DynamoDBIntegrationLoadSpec
expectMsgAllOf(s"a-${i}", i, false)
}
- val testProbe = TestProbe()
+ val testProbe = TestProbe()
val processor2 = system.actorOf(Props(classOf[ProcessorAtomic], persistenceId, testProbe.ref))
(1L to 6L).foreach { i =>
testProbe.expectMsgAllOf(s"a-${i}", i, true)
}
}
"write and replay with persistAll greater than partition size skipping part of a partition" in {
- val persistenceId = UUID.randomUUID().toString
- val probe = TestProbe()
+ val persistenceId = UUID.randomUUID().toString
+ val probe = TestProbe()
val processorAtomic = system.actorOf(Props(classOf[ProcessorAtomic], persistenceId, self))
processorAtomic ! List("a-1", "a-2", "a-3")
@@ -248,15 +248,15 @@ class DynamoDBIntegrationLoadSpec
expectMsgAllOf(s"a-${i}", i, false)
}
- val testProbe = TestProbe()
+ val testProbe = TestProbe()
val processor2 = system.actorOf(Props(classOf[ProcessorAtomic], persistenceId, testProbe.ref))
(1L to 6L).foreach { i =>
testProbe.expectMsgAllOf(s"a-${i}", i, true)
}
}
"write and replay with persistAll less than partition size" in {
- val persistenceId = UUID.randomUUID().toString
- val probe = TestProbe()
+ val persistenceId = UUID.randomUUID().toString
+ val probe = TestProbe()
val processorAtomic = system.actorOf(Props(classOf[ProcessorAtomic], persistenceId, self))
processorAtomic ! List("a-1", "a-2", "a-3", "a-4")
@@ -271,8 +271,8 @@ class DynamoDBIntegrationLoadSpec
}
"not replay messages deleted from the +1 partition" in {
val persistenceId = UUID.randomUUID().toString
- val probe = TestProbe()
- val deleteProbe = TestProbe()
+ val probe = TestProbe()
+ val deleteProbe = TestProbe()
subscribeToRangeDeletion(deleteProbe)
val processorAtomic = system.actorOf(Props(classOf[ProcessorAtomic], persistenceId, self))
@@ -283,7 +283,7 @@ class DynamoDBIntegrationLoadSpec
processorAtomic ! DeleteTo(5L)
awaitRangeDeletion(deleteProbe)
- val testProbe = TestProbe()
+ val testProbe = TestProbe()
val processor2 = system.actorOf(Props(classOf[ProcessorAtomic], persistenceId, testProbe.ref))
testProbe.expectMsgAllOf(s"a-6", 6, true)
}
diff --git a/src/test/scala/akka/persistence/dynamodb/journal/DynamoDBUtils.scala b/src/test/scala/akka/persistence/dynamodb/journal/DynamoDBUtils.scala
index 8c60635..427c84d 100644
--- a/src/test/scala/akka/persistence/dynamodb/journal/DynamoDBUtils.scala
+++ b/src/test/scala/akka/persistence/dynamodb/journal/DynamoDBUtils.scala
@@ -20,7 +20,7 @@ trait DynamoDBUtils extends JournalSettingsProvider with DynamoProvider {
import system.dispatcher
override lazy val journalSettings = {
- val c = system.settings.config
+ val c = system.settings.config
val config = c.getConfig(c.getString("akka.persistence.journal.plugin"))
new DynamoDBJournalConfig(config)
}
@@ -56,7 +56,7 @@ trait DynamoDBUtils extends JournalSettingsProvider with DynamoProvider {
Await.result(setup, 5.seconds)
}
- private val writerUuid = UUID.randomUUID.toString
+ private val writerUuid = UUID.randomUUID.toString
def persistenceId: String = ???
var nextSeqNr = 1
diff --git a/src/test/scala/akka/persistence/dynamodb/journal/FailureReportingSpec.scala b/src/test/scala/akka/persistence/dynamodb/journal/FailureReportingSpec.scala
index a27ed64..712fc23 100644
--- a/src/test/scala/akka/persistence/dynamodb/journal/FailureReportingSpec.scala
+++ b/src/test/scala/akka/persistence/dynamodb/journal/FailureReportingSpec.scala
@@ -68,8 +68,9 @@ class FailureReportingSpec
.withFallback(ConfigFactory.load())
implicit val system = ActorSystem("FailureReportingSpec-test1", config)
try EventFilter[ResourceNotFoundException](pattern = ".*ThisTableDoesNotExist.*", occurrences = 1).intercept {
- Persistence(system).journalFor("")
- } finally system.terminate()
+ Persistence(system).journalFor("")
+ }
+ finally system.terminate()
}
"keep running even when journal table is absent" in {
@@ -101,12 +102,13 @@ class FailureReportingSpec
.withFallback(ConfigFactory.load())
implicit val system = ActorSystem("FailureReportingSpec-test3", config)
try EventFilter.info(pattern = ".*protocol:https.*", occurrences = 1).intercept {
- Persistence(system).journalFor("")
- } finally system.terminate()
+ Persistence(system).journalFor("")
+ }
+ finally system.terminate()
}
"not notify user about config errors when starting the default journal" in {
- val config = ConfigFactory.parseString("""
+ val config = ConfigFactory.parseString("""
dynamodb-journal {
endpoint = "http://localhost:8888"
aws-access-key-id = "AWS_ACCESS_KEY_ID"
@@ -161,7 +163,7 @@ akka.loggers = ["akka.testkit.TestEventListener"]
"properly reject too large payloads" in {
val journal = Persistence(system).journalFor("")
- val msgs = Vector("t-1", bigMsg, "t-3", "t-4", bigMsg, "t-6").map(persistentRepr)
+ val msgs = Vector("t-1", bigMsg, "t-3", "t-4", bigMsg, "t-6").map(persistentRepr)
val write =
AtomicWrite(msgs(0)) ::
AtomicWrite(msgs(1)) ::
@@ -192,7 +194,7 @@ akka.loggers = ["akka.testkit.TestEventListener"]
import dynamo._
def desc[T](aws: T)(implicit d: Describe[_ >: T]): String = d.desc(aws)
- val keyItem = Map(Key -> S("TheKey"), Sort -> N("42")).asJava
+ val keyItem = Map(Key -> S("TheKey"), Sort -> N("42")).asJava
val key2Item = Map(Key -> S("The2Key"), Sort -> N("43")).asJava
"reporting table problems" in {
@@ -226,9 +228,9 @@ akka.loggers = ["akka.testkit.TestEventListener"]
}
"reporting batch write problems" in {
- val write = new WriteRequest().withPutRequest(new PutRequest().withItem(keyItem))
+ val write = new WriteRequest().withPutRequest(new PutRequest().withItem(keyItem))
val remove = new WriteRequest().withDeleteRequest(new DeleteRequest().withKey(key2Item))
- val aws = new BatchWriteItemRequest().withRequestItems(Map("TheTable" -> Seq(write, remove).asJava).asJava)
+ val aws = new BatchWriteItemRequest().withRequestItems(Map("TheTable" -> Seq(write, remove).asJava).asJava)
desc(aws) should include("BatchWriteItem")
desc(aws) should include("TheTable")
desc(aws) should include("put[par=TheKey,num=42]")
@@ -236,7 +238,7 @@ akka.loggers = ["akka.testkit.TestEventListener"]
}
"reporting batch read problems" in {
- val ka = new KeysAndAttributes().withKeys(keyItem, key2Item)
+ val ka = new KeysAndAttributes().withKeys(keyItem, key2Item)
val aws = new BatchGetItemRequest().withRequestItems(Map("TheTable" -> ka).asJava)
desc(aws) should include("BatchGetItem")
desc(aws) should include("TheTable")
diff --git a/src/test/scala/akka/persistence/dynamodb/journal/PartialAsyncSerializationSpec.scala b/src/test/scala/akka/persistence/dynamodb/journal/PartialAsyncSerializationSpec.scala
index eefe3aa..c4d62ac 100644
--- a/src/test/scala/akka/persistence/dynamodb/journal/PartialAsyncSerializationSpec.scala
+++ b/src/test/scala/akka/persistence/dynamodb/journal/PartialAsyncSerializationSpec.scala
@@ -54,7 +54,7 @@ class PartialAsyncSerializationSpec
}
override val persistenceId = "PartialAsyncSerializationSpec"
- lazy val journal = Persistence(system).journalFor("")
+ lazy val journal = Persistence(system).journalFor("")
"DynamoDB Journal (Serialization Test)" must {
diff --git a/src/test/scala/akka/persistence/dynamodb/journal/PersistAllConsistencySpec.scala b/src/test/scala/akka/persistence/dynamodb/journal/PersistAllConsistencySpec.scala
index 7d93542..8818466 100644
--- a/src/test/scala/akka/persistence/dynamodb/journal/PersistAllConsistencySpec.scala
+++ b/src/test/scala/akka/persistence/dynamodb/journal/PersistAllConsistencySpec.scala
@@ -35,7 +35,7 @@ class PersistAllConsistencySpec
}
override val persistenceId = "PersistAllConsistencySpec"
- lazy val journal = Persistence(system).journalFor("")
+ lazy val journal = Persistence(system).journalFor("")
"DynamoDB Journal (persistAll)" must {
@@ -44,7 +44,7 @@ class PersistAllConsistencySpec
expectMsg(Purged(persistenceId))
val start = nextSeqNr
- val end = 10
+ val end = 10
println(s"start: ${start}; end: ${end}")
val padding = AtomicWrite((start to end).map(i => persistentRepr(f"h-$i"))) :: Nil
@@ -60,7 +60,7 @@ class PersistAllConsistencySpec
for (t <- Seq(("last", 3), ("middle", 2), ("first", 1)))
s"correctly cross page boundaries with AtomicWrite position ${t._1}" in {
val start1 = nextSeqNr
- val end1 = ((start1 / PartitionSize) + 1) * PartitionSize - t._2
+ val end1 = ((start1 / PartitionSize) + 1) * PartitionSize - t._2
println(s"start: ${start1}; end: ${end1}")
val padding = AtomicWrite((start1 to end1).map(i => persistentRepr(f"h-$i"))) :: Nil
@@ -69,7 +69,7 @@ class PersistAllConsistencySpec
(start1 to end1).foreach(i => expectMsg(WriteMessageSuccess(generatedMessages(i), 1)))
val start2 = nextSeqNr
- val end2 = start2 + 2
+ val end2 = start2 + 2
println(s"start: ${start2}; end: ${end2}")
val subject = AtomicWrite((start2 to end2).map(i => persistentRepr(f"h-$i"))) :: Nil
@@ -84,7 +84,7 @@ class PersistAllConsistencySpec
s"recover correctly when the last partition event ends on ${PartitionSize - 1}" in {
val start = nextSeqNr
- val end = ((start / PartitionSize) + 1) * PartitionSize - 1
+ val end = ((start / PartitionSize) + 1) * PartitionSize - 1
println(s"start: ${start}; end: ${end}")
val padding = AtomicWrite((start to end).map(i => persistentRepr(f"h-$i"))) :: Nil
diff --git a/src/test/scala/akka/persistence/dynamodb/journal/RecoveryConsistencySpec.scala b/src/test/scala/akka/persistence/dynamodb/journal/RecoveryConsistencySpec.scala
index 5b8ef00..860d583 100644
--- a/src/test/scala/akka/persistence/dynamodb/journal/RecoveryConsistencySpec.scala
+++ b/src/test/scala/akka/persistence/dynamodb/journal/RecoveryConsistencySpec.scala
@@ -44,7 +44,7 @@ class RecoveryConsistencySpec
super.afterAll()
}
- override val persistenceId = "RecoveryConsistencySpec"
+ override val persistenceId = "RecoveryConsistencySpec"
implicit val materializer: Materializer = SystemMaterializer(system).materializer
private lazy val journal = Persistence(system).journalFor("")
@@ -54,11 +54,11 @@ class RecoveryConsistencySpec
"DynamoDB Journal (Recovery)" must {
- val repetitions = 50
+ val repetitions = 50
val nrOfMessages = 20
- val messages = (1 to nrOfMessages).map(i => f"a-$i%04d")
- val writes = messages.map(m => AtomicWrite(persistentRepr(m)))
- val probe = TestProbe()
+ val messages = (1 to nrOfMessages).map(i => f"a-$i%04d")
+ val writes = messages.map(m => AtomicWrite(persistentRepr(m)))
+ val probe = TestProbe()
for (i <- 1 to repetitions)
s"not return intermediate values for the highest sequence number ($i of $repetitions)" in {
@@ -78,9 +78,9 @@ class RecoveryConsistencySpec
"only replay completely persisted AtomicWrites" in {
val more =
- AtomicWrite((1 to 3).map(i => persistentRepr(f"b-$i"))) :: // hole in the middle
- AtomicWrite((4 to 6).map(i => persistentRepr(f"b-$i"))) :: // hole in the beginning
- AtomicWrite((7 to 9).map(i => persistentRepr(f"b-$i"))) :: // no hole
+ AtomicWrite((1 to 3).map(i => persistentRepr(f"b-$i"))) :: // hole in the middle
+ AtomicWrite((4 to 6).map(i => persistentRepr(f"b-$i"))) :: // hole in the beginning
+ AtomicWrite((7 to 9).map(i => persistentRepr(f"b-$i"))) :: // no hole
AtomicWrite((10 to 12).map(i => persistentRepr(f"b-$i"))) :: // hole in the end
AtomicWrite((13 to 15).map(i => persistentRepr(f"b-$i"))) :: // hole in the end
AtomicWrite(persistentRepr("c")) ::
@@ -105,8 +105,8 @@ class RecoveryConsistencySpec
"read correct highest sequence number even if a Sort=0 entry is lost" in {
val start = nrOfMessages + 19
- val end = (start / PartitionSize + 1) * PartitionSize
- val more = (start to end).map(i => AtomicWrite(persistentRepr(f"e-$i")))
+ val end = (start / PartitionSize + 1) * PartitionSize
+ val more = (start to end).map(i => AtomicWrite(persistentRepr(f"e-$i")))
journal ! WriteMessages(more, testActor, 1)
expectMsg(WriteMessagesSuccessful)
(start to end).foreach(i => expectMsg(WriteMessageSuccess(generatedMessages(i), 1)))
diff --git a/src/test/scala/akka/persistence/dynamodb/journal/WriteThroughputBench.scala b/src/test/scala/akka/persistence/dynamodb/journal/WriteThroughputBench.scala
index 865135f..98226f1 100644
--- a/src/test/scala/akka/persistence/dynamodb/journal/WriteThroughputBench.scala
+++ b/src/test/scala/akka/persistence/dynamodb/journal/WriteThroughputBench.scala
@@ -17,7 +17,7 @@ import scala.concurrent.duration._
object WriteThroughputBench extends App with DynamoDBUtils {
- def rnd = ThreadLocalRandom.current()
+ def rnd = ThreadLocalRandom.current()
final val oneBillion = 1000L * 1000 * 1000
class H private (private val entries: Map[Int, Int]) {
@@ -45,8 +45,8 @@ object WriteThroughputBench extends App with DynamoDBUtils {
}
}
object Report {
- def apply(): Report = Report(new Histogram(3), new Histogram(3), new H)
- def endToEnd(h: Histogram): Report = Report(h, new Histogram(3), new H)
+ def apply(): Report = Report(new Histogram(3), new Histogram(3), new H)
+ def endToEnd(h: Histogram): Report = Report(h, new Histogram(3), new H)
def calls(h: Histogram, r: H): Report = Report(new Histogram(3), h, r)
}
@@ -54,8 +54,8 @@ object WriteThroughputBench extends App with DynamoDBUtils {
val event = new Array[Byte](100)
rnd.nextBytes(event)
- var nextReport = System.nanoTime + oneBillion
- val histo = new Histogram(3)
+ var nextReport = System.nanoTime + oneBillion
+ val histo = new Histogram(3)
val persistenceId = UUID.randomUUID().toString
self ! Go
@@ -116,7 +116,7 @@ writer-dispatcher {
""").resolve)
.withFallback(ConfigFactory.load())
- implicit val system = ActorSystem("WriteThroughputBench", config)
+ implicit val system = ActorSystem("WriteThroughputBench", config)
implicit val materializer = ActorMaterializer(ActorMaterializerSettings(system).withInputBuffer(1, 1))
/*
@@ -152,15 +152,15 @@ writer-dispatcher {
.fromGraph(GraphDSL.create(endToEnd, calls)(Keep.both) { implicit b => (e, c) =>
val zip = b.add(ZipWith((_: Unit, er: Report, cr: Report) => er + cr))
Source.tick(1.second, 1.second, ()) ~> zip.in0
- e ~> zip.in1
- c ~> zip.in2
- zip.out ~> Sink.foreach(printStats)
+ e ~> zip.in1
+ c ~> zip.in2
+ zip.out ~> Sink.foreach(printStats)
ClosedShape
})
.run()
println(s"starting $writers writers")
- val props = Props(new Writer(eRef)).withDispatcher("writer-dispatcher")
+ val props = Props(new Writer(eRef)).withDispatcher("writer-dispatcher")
val writerRefs = (1 to writers).map(_ => system.actorOf(props))
Persistence(system).journalFor("") ! SetDBHelperReporter(cRef)
@@ -176,6 +176,6 @@ writer-dispatcher {
def perc(h: Histogram) =
f"50=${p(h, 0.5)}%5.1f 90=${p(h, 0.9)}%5.1f 99=${p(h, 0.99)}%5.1f 99.9=${p(h, 0.999)}%5.1f 99.99=${p(h, 0.9999)}%5.1f"
println(f"count ${r.endToEnd.getTotalCount}%6d/s percentiles: endToEnd(${perc(r.endToEnd)}) calls(${perc(
- r.calls)}) retries: ${r.retries}")
+ r.calls)}) retries: ${r.retries}")
}
}
diff --git a/src/test/scala/akka/persistence/dynamodb/query/CurrentPersistenceIdsSpec.scala b/src/test/scala/akka/persistence/dynamodb/query/CurrentPersistenceIdsSpec.scala
index ace01e1..17517e5 100644
--- a/src/test/scala/akka/persistence/dynamodb/query/CurrentPersistenceIdsSpec.scala
+++ b/src/test/scala/akka/persistence/dynamodb/query/CurrentPersistenceIdsSpec.scala
@@ -32,11 +32,11 @@ class CurrentPersistenceIdsSpec
with ReadJournalSettingsProvider
with DynamoProvider {
override protected lazy val readJournalSettings: DynamoDBReadJournalConfig = DynamoDBReadJournalConfig()
- override implicit val patienceConfig: PatienceConfig = PatienceConfig(15.seconds)
+ override implicit val patienceConfig: PatienceConfig = PatienceConfig(15.seconds)
- private val writerUuid = UUID.randomUUID.toString
+ private val writerUuid = UUID.randomUUID.toString
private implicit val materializer: Materializer = SystemMaterializer(system).materializer
- private lazy val journal = Persistence(system).journalFor("")
+ private lazy val journal = Persistence(system).journalFor("")
private lazy val queries =
PersistenceQuery(system).readJournalFor[DynamoDBReadJournal](DynamoDBReadJournal.Identifier)
@@ -69,16 +69,14 @@ class CurrentPersistenceIdsSpec
private def persistEvents(persistenceIds: Seq[String]): Unit = {
val eventsPerActor = 0 to 5
- val writes = persistenceIds.map(
- persistenceId =>
- AtomicWrite(
- eventsPerActor.map(
- i =>
- PersistentRepr(
- payload = s"$persistenceId $i",
- sequenceNr = i,
- persistenceId = persistenceId,
- writerUuid = writerUuid))))
+ val writes = persistenceIds.map(persistenceId =>
+ AtomicWrite(
+ eventsPerActor.map(i =>
+ PersistentRepr(
+ payload = s"$persistenceId $i",
+ sequenceNr = i,
+ persistenceId = persistenceId,
+ writerUuid = writerUuid))))
writes.foreach { message =>
journal ! WriteMessages(message :: Nil, testActor, 1)
diff --git a/src/test/scala/akka/persistence/dynamodb/snapshot/DynamoDBUtils.scala b/src/test/scala/akka/persistence/dynamodb/snapshot/DynamoDBUtils.scala
index 396baab..9acd40d 100644
--- a/src/test/scala/akka/persistence/dynamodb/snapshot/DynamoDBUtils.scala
+++ b/src/test/scala/akka/persistence/dynamodb/snapshot/DynamoDBUtils.scala
@@ -23,7 +23,7 @@ trait DynamoDBUtils {
implicit val executionContext = system.dispatcher
lazy val settings: DynamoDBSnapshotConfig = {
- val c = system.settings.config
+ val c = system.settings.config
val config = c.getConfig(c.getString("akka.persistence.snapshot-store.plugin"))
new DynamoDBSnapshotConfig(config)
}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pekko.apache.org
For additional commands, e-mail: commits-help@pekko.apache.org
[incubator-pekko-persistence-dynamodb] 02/03: Add .gitattributes to enforce unix line endings
Posted by md...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
mdedetrich pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/incubator-pekko-persistence-dynamodb.git
commit eb7a1cdd03b1024b0c7286aae0601b791daab619
Author: Matthew de Detrich <ma...@aiven.io>
AuthorDate: Fri Nov 11 18:01:45 2022 +0100
Add .gitattributes to enforce unix line endings
---
.gitattributes | 5 +++++
1 file changed, 5 insertions(+)
diff --git a/.gitattributes b/.gitattributes
new file mode 100644
index 0000000..9dde9b9
--- /dev/null
+++ b/.gitattributes
@@ -0,0 +1,5 @@
+# Activate line ending normalization, setting eol will make the behavior match core.autocrlf = input
+* text=auto eol=lf
+# Force batch scripts to always use CRLF line endings
+*.{cmd,[cC][mM][dD]} text eol=crlf
+*.{bat,[bB][aA][tT]} text eol=crlf
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pekko.apache.org
For additional commands, e-mail: commits-help@pekko.apache.org