You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pekko.apache.org by fa...@apache.org on 2023/12/20 07:32:59 UTC
(incubator-pekko-persistence-dynamodb) branch main updated: changes due to pekko deprecations (#90)
This is an automated email from the ASF dual-hosted git repository.
fanningpj pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/incubator-pekko-persistence-dynamodb.git
The following commit(s) were added to refs/heads/main by this push:
new 268967f changes due to pekko deprecations (#90)
268967f is described below
commit 268967f01a3f629f96f46c5e3ab2ac4860d18c38
Author: PJ Fanning <pj...@users.noreply.github.com>
AuthorDate: Wed Dec 20 08:32:54 2023 +0100
changes due to pekko deprecations (#90)
---
.../apache/pekko/persistence/dynamodb/journal/DynamoDBJournal.scala | 2 +-
.../persistence/dynamodb/journal/DynamoDBJournalRequests.scala | 2 +-
.../pekko/persistence/dynamodb/journal/DynamoDBRecovery.scala | 2 +-
src/main/scala/org/apache/pekko/persistence/dynamodb/package.scala | 4 ++--
.../internal/DynamoDBCurrentEventsByPersistenceIdQuery.scala | 2 +-
.../scaladsl/internal/DynamoDBCurrentPersistenceIdsQuery.scala | 2 +-
.../pekko/persistence/dynamodb/journal/WriteThroughputBench.scala | 2 +-
.../persistence/dynamodb/journal/DynamoDBIntegrationLoadSpec.scala | 6 +++---
.../pekko/persistence/dynamodb/journal/FailureReportingSpec.scala | 2 +-
9 files changed, 12 insertions(+), 12 deletions(-)
diff --git a/src/main/scala/org/apache/pekko/persistence/dynamodb/journal/DynamoDBJournal.scala b/src/main/scala/org/apache/pekko/persistence/dynamodb/journal/DynamoDBJournal.scala
index 3ab1be7..c62877c 100644
--- a/src/main/scala/org/apache/pekko/persistence/dynamodb/journal/DynamoDBJournal.scala
+++ b/src/main/scala/org/apache/pekko/persistence/dynamodb/journal/DynamoDBJournal.scala
@@ -118,7 +118,7 @@ class DynamoDBJournal(config: Config)
f.onComplete { _ =>
self ! OpFinished(pid, p.future)
p.success(Done)
- }(ExecutionContexts.sameThreadExecutionContext)
+ }(ExecutionContexts.parasitic)
f
}
diff --git a/src/main/scala/org/apache/pekko/persistence/dynamodb/journal/DynamoDBJournalRequests.scala b/src/main/scala/org/apache/pekko/persistence/dynamodb/journal/DynamoDBJournalRequests.scala
index 137a89b..7175f2b 100644
--- a/src/main/scala/org/apache/pekko/persistence/dynamodb/journal/DynamoDBJournalRequests.scala
+++ b/src/main/scala/org/apache/pekko/persistence/dynamodb/journal/DynamoDBJournalRequests.scala
@@ -47,7 +47,7 @@ trait DynamoDBJournalRequests extends DynamoDBRequests {
// optimize the common case
if (writes.size == 1) {
writeMessages(writes.head)
- .map(bubbleUpFailures(_) :: Nil)(ExecutionContexts.sameThreadExecutionContext)
+ .map(bubbleUpFailures(_) :: Nil)(ExecutionContexts.parasitic)
} else {
def rec(todo: List[AtomicWrite], acc: List[Try[Unit]]): Future[List[Try[Unit]]] =
todo match {
diff --git a/src/main/scala/org/apache/pekko/persistence/dynamodb/journal/DynamoDBRecovery.scala b/src/main/scala/org/apache/pekko/persistence/dynamodb/journal/DynamoDBRecovery.scala
index d348d17..f80d88a 100644
--- a/src/main/scala/org/apache/pekko/persistence/dynamodb/journal/DynamoDBRecovery.scala
+++ b/src/main/scala/org/apache/pekko/persistence/dynamodb/journal/DynamoDBRecovery.scala
@@ -233,7 +233,7 @@ trait DynamoDBRecovery extends AsyncReplayMessages {
toSequenceNr: Long,
max: Long): Source[PersistentRepr, NotUsed] =
// toSequenceNr is already capped to highest and guaranteed to be no less than fromSequenceNr
- Source.fromFuture(readSequenceNr(persistenceId, highest = false)).flatMapConcat { lowest =>
+ Source.future(readSequenceNr(persistenceId, highest = false)).flatMapConcat { lowest =>
val start = Math.max(fromSequenceNr, lowest)
val async = ReplayParallelism > 1
diff --git a/src/main/scala/org/apache/pekko/persistence/dynamodb/package.scala b/src/main/scala/org/apache/pekko/persistence/dynamodb/package.scala
index ca2cdc0..76f04c6 100644
--- a/src/main/scala/org/apache/pekko/persistence/dynamodb/package.scala
+++ b/src/main/scala/org/apache/pekko/persistence/dynamodb/package.scala
@@ -42,7 +42,7 @@ package object dynamodb {
def lift[T](f: Future[T]): Future[Try[T]] = {
val p = Promise[Try[T]]()
- f.onComplete(p.success)(ExecutionContexts.sameThreadExecutionContext)
+ f.onComplete(p.success)(ExecutionContexts.parasitic)
p.future
}
@@ -51,7 +51,7 @@ package object dynamodb {
f.onComplete {
case Success(_) => p.success(Success(()))
case f @ Failure(_) => p.success(f.asInstanceOf[Failure[Unit]])
- }(ExecutionContexts.sameThreadExecutionContext)
+ }(ExecutionContexts.parasitic)
p.future
}
diff --git a/src/main/scala/org/apache/pekko/persistence/dynamodb/query/scaladsl/internal/DynamoDBCurrentEventsByPersistenceIdQuery.scala b/src/main/scala/org/apache/pekko/persistence/dynamodb/query/scaladsl/internal/DynamoDBCurrentEventsByPersistenceIdQuery.scala
index 0e74333..ba55532 100644
--- a/src/main/scala/org/apache/pekko/persistence/dynamodb/query/scaladsl/internal/DynamoDBCurrentEventsByPersistenceIdQuery.scala
+++ b/src/main/scala/org/apache/pekko/persistence/dynamodb/query/scaladsl/internal/DynamoDBCurrentEventsByPersistenceIdQuery.scala
@@ -57,7 +57,7 @@ trait DynamoDBCurrentEventsByPersistenceIdQuery
require(fromSequenceNr < toSequenceNr, "fromSequenceNr should be smaller than toSequenceNr")
log.debug("starting currentEventsByPersistenceId for {} from {} to {}", persistenceId, fromSequenceNr, toSequenceNr)
Source
- .fromFuture(readSequenceNr(persistenceId = persistenceId, highest = true))
+ .future(readSequenceNr(persistenceId = persistenceId, highest = true))
.flatMapConcat { highest =>
val end = Math.min(highest, toSequenceNr)
eventsStream(
diff --git a/src/main/scala/org/apache/pekko/persistence/dynamodb/query/scaladsl/internal/DynamoDBCurrentPersistenceIdsQuery.scala b/src/main/scala/org/apache/pekko/persistence/dynamodb/query/scaladsl/internal/DynamoDBCurrentPersistenceIdsQuery.scala
index 2f5c8af..5920da0 100644
--- a/src/main/scala/org/apache/pekko/persistence/dynamodb/query/scaladsl/internal/DynamoDBCurrentPersistenceIdsQuery.scala
+++ b/src/main/scala/org/apache/pekko/persistence/dynamodb/query/scaladsl/internal/DynamoDBCurrentPersistenceIdsQuery.scala
@@ -114,7 +114,7 @@ trait DynamoDBCurrentPersistenceIdsQuery extends PublicDynamoDBCurrentPersistenc
}
val infiniteStreamOfResults: ResultSource =
- lazyStream(Source.fromFuture(getPersistenceIds(None).map(Some(_))))
+ lazyStream(Source.future(getPersistenceIds(None).map(Some(_))))
infiniteStreamOfResults
.takeWhile(_.isDefined)
diff --git a/src/test/scala-2/org/apache/pekko/persistence/dynamodb/journal/WriteThroughputBench.scala b/src/test/scala-2/org/apache/pekko/persistence/dynamodb/journal/WriteThroughputBench.scala
index 8f74495..aabadfd 100644
--- a/src/test/scala-2/org/apache/pekko/persistence/dynamodb/journal/WriteThroughputBench.scala
+++ b/src/test/scala-2/org/apache/pekko/persistence/dynamodb/journal/WriteThroughputBench.scala
@@ -160,7 +160,7 @@ writer-dispatcher {
val (eRef, cRef) =
RunnableGraph
- .fromGraph(GraphDSL.create(endToEnd, calls)(Keep.both) { implicit b => (e, c) =>
+ .fromGraph(GraphDSL.createGraph(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
diff --git a/src/test/scala/org/apache/pekko/persistence/dynamodb/journal/DynamoDBIntegrationLoadSpec.scala b/src/test/scala/org/apache/pekko/persistence/dynamodb/journal/DynamoDBIntegrationLoadSpec.scala
index dd65576..82e0f55 100644
--- a/src/test/scala/org/apache/pekko/persistence/dynamodb/journal/DynamoDBIntegrationLoadSpec.scala
+++ b/src/test/scala/org/apache/pekko/persistence/dynamodb/journal/DynamoDBIntegrationLoadSpec.scala
@@ -215,19 +215,19 @@ class DynamoDBIntegrationLoadSpec
}
val view = system.actorOf(Props(classOf[ViewA], "p7-view", persistenceId, probe.ref))
- probe.expectNoMsg(200.millis)
+ probe.expectNoMessage(200.millis)
view ! Update(await = true, replayMax = 3L)
probe.expectMsg(s"a-1")
probe.expectMsg(s"a-2")
probe.expectMsg(s"a-3")
- probe.expectNoMsg(200.millis)
+ probe.expectNoMessage(200.millis)
view ! Update(await = true, replayMax = 3L)
probe.expectMsg(s"a-4")
probe.expectMsg(s"a-5")
probe.expectMsg(s"a-6")
- probe.expectNoMsg(200.millis)
+ probe.expectNoMessage(200.millis)
} */
"write and replay with persistAll greater than partition size skipping whole partition" in {
val persistenceId = UUID.randomUUID().toString
diff --git a/src/test/scala/org/apache/pekko/persistence/dynamodb/journal/FailureReportingSpec.scala b/src/test/scala/org/apache/pekko/persistence/dynamodb/journal/FailureReportingSpec.scala
index 4d5e70d..1f7bb7e 100644
--- a/src/test/scala/org/apache/pekko/persistence/dynamodb/journal/FailureReportingSpec.scala
+++ b/src/test/scala/org/apache/pekko/persistence/dynamodb/journal/FailureReportingSpec.scala
@@ -140,7 +140,7 @@ pekko.loggers = ["org.apache.pekko.testkit.TestEventListener"]
}
probe.expectMsgType[Logging.Error].message.toString should include("DescribeTableRequest(pekko-persistence)")
probe.expectMsgType[Logging.Error].message.toString should include("until the table 'pekko-persistence'")
- probe.expectNoMsg(0.seconds)
+ probe.expectNoMessage(0.seconds)
} finally system.terminate()
}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pekko.apache.org
For additional commands, e-mail: commits-help@pekko.apache.org