Skip to content

Commit c627a6d

Browse files
author
Dennis Vriend
committed
refactoring
1 parent 24d5597 commit c627a6d

File tree

8 files changed

+118
-145
lines changed

8 files changed

+118
-145
lines changed

‎.gitignore

+3-2
Original file line numberDiff line numberDiff line change
@@ -1,4 +1,5 @@
1-
.vagrant/
1+
project/project/
2+
project/target/
23
target/
3-
.idea
4+
.idea/
45
*.iml

‎.travis.yml

+13
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,13 @@
1+
language: scala
2+
sudo: false
3+
scala:
4+
- "2.10.5"
5+
- "2.11.6"
6+
jdk:
7+
- oraclejdk8
8+
branches:
9+
only:
10+
- master
11+
notifications:
12+
email:
13+
- [email protected]

‎README.md

+10-1
Original file line numberDiff line numberDiff line change
@@ -2,12 +2,14 @@
22
Akka-persistence-inmemory is a plugin for [akka-persistence](http://doc.akka.io/docs/akka/snapshot/scala/persistence.html)
33
that writes journal and snapshot entries entries to an in-memory store. It is very useful for testing your persistent actors.
44

5+
[![Build Status](https://travis-ci.org/dnvriend/akka-persistence-in-memory.svg?branch=master)](https://travis-ci.org/dnvriend/akka-persistence-in-memory)
6+
57
# Dependency
68
To include the plugin into your sbt project, add the following lines to your build.sbt file:
79

810
resolvers += "dnvriend at bintray" at "http://dl.bintray.com/dnvriend/maven"
911

10-
libraryDependencies += "com.github.dnvriend" %% "akka-persistence-inmemory" % "1.0.0"
12+
libraryDependencies += "com.github.dnvriend" %% "akka-persistence-inmemory" % "1.0.1"
1113

1214
# Configuration
1315
Add the following to the application.conf:
@@ -23,6 +25,13 @@ akka {
2325

2426
## What's new?
2527

28+
## 1.0.1 (2015-05-16)
29+
- Some refactoring, fixed some misconceptions about the behavior of Scala Futures one year ago :)
30+
- Akka 2.3.6 -> 2.3.11
31+
- Scala 2.11.1 -> 2.11.6
32+
- Scala 2.10.4 -> 2.10.5
33+
- Merged Issue #1 (Sebastián Ortega)[https://github.com/sortega] Fix corner case when persisted events are deleted, thanks!
34+
2635
## 1.0.0
2736
- Moved to bintray
2837

‎bintray.sbt

+8
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,8 @@
1+
import bintray.Plugin._
2+
3+
seq(bintraySettings:_*)
4+
5+
bintray.Keys.packageLabels in bintray.Keys.bintray := Seq("akka", "persistence", "in-memory")
6+
7+
bintray.Keys.packageAttributes in bintray.Keys.bintray ~=
8+
((_: bintray.AttrMap) ++ Map("website_url" -> Seq(bintry.StringAttr("https://github.com/dnvriend/akka-persistence-inmemory.git")), "github_repo" -> Seq(bintry.StringAttr("https://github.com/dnvriend/akka-persistence-inmemory.git")), "issue_tracker_url" -> Seq(bintry.StringAttr("https://github.com/dnvriend/akka-persistence-inmemory.git/issues/"))))

‎build.sbt

+5-12
Original file line numberDiff line numberDiff line change
@@ -1,21 +1,17 @@
1-
import bintray.Plugin._
2-
3-
seq(bintraySettings:_*)
4-
51
organization := "com.github.dnvriend"
62

73
name := "akka-persistence-inmemory"
84

9-
version := "1.0.0"
5+
version := "1.0.1-SNAPSHOT"
106

11-
scalaVersion := "2.11.1"
7+
scalaVersion := "2.11.6"
128

13-
crossScalaVersions := Seq("2.10.4", "2.11.2")
9+
crossScalaVersions := Seq("2.10.5", "2.11.6")
1410

1511
resolvers += "krasserm at bintray" at "http://dl.bintray.com/krasserm/maven"
1612

1713
libraryDependencies ++= {
18-
val akkaVersion = "2.3.6"
14+
val akkaVersion = "2.3.11"
1915
Seq(
2016
"com.typesafe.akka" %% "akka-actor" % akkaVersion,
2117
"com.typesafe.akka" %% "akka-slf4j" % akkaVersion,
@@ -34,7 +30,4 @@ publishMavenStyle := true
3430

3531
licenses += ("Apache-2.0", url("http://opensource.org/licenses/apache2.0.php"))
3632

37-
bintray.Keys.packageLabels in bintray.Keys.bintray := Seq("akka", "elasticsearch", "lucene", "cluster", "index", "indexing")
38-
39-
bintray.Keys.packageAttributes in bintray.Keys.bintray ~=
40-
((_: bintray.AttrMap) ++ Map("website_url" -> Seq(bintry.StringAttr("https://github.com/dnvriend/akka-persistence-inmemory.git")), "github_repo" -> Seq(bintry.StringAttr("https://github.com/dnvriend/akka-persistence-inmemory.git")), "issue_tracker_url" -> Seq(bintry.StringAttr("https://github.com/dnvriend/akka-persistence-inmemory.git/issues/"))))
33+
parallelExecution in Test := false

‎project/build.properties

+1-1
Original file line numberDiff line numberDiff line change
@@ -1 +1 @@
1-
sbt.version=0.13.5
1+
sbt.version=0.13.8

‎src/main/scala/akka/persistence/inmemory/journal/InMemoryJournal.scala

+58-96
Original file line numberDiff line numberDiff line change
@@ -1,10 +1,11 @@
11
package akka.persistence.inmemory.journal
22

3+
import java.util.concurrent.ConcurrentHashMap
4+
35
import akka.actor.ActorLogging
4-
import akka.persistence.{PersistentConfirmation, PersistentId, PersistentRepr}
56
import akka.persistence.journal.AsyncWriteJournal
7+
import akka.persistence.{PersistentConfirmation, PersistentId, PersistentRepr}
68

7-
import java.util.concurrent.ConcurrentHashMap
89
import scala.collection.JavaConverters._
910
import scala.collection.immutable.Seq
1011
import scala.concurrent.Future
@@ -13,122 +14,83 @@ class InMemoryJournal extends AsyncWriteJournal with ActorLogging {
1314
implicit val ec = context.system.dispatcher
1415
val journal: scala.collection.mutable.Map[String, List[PersistentRepr]] = new ConcurrentHashMap[String, List[PersistentRepr]].asScala
1516

16-
override def asyncWriteMessages(messages: Seq[PersistentRepr]): Future[Unit] = {
17-
Future[Unit] {
18-
val mess = messages
19-
log.debug("writeMessages for {} persistent messages", mess.size)
20-
mess.foreach { repr =>
21-
import repr._
22-
journal.get(persistenceId) match {
23-
case None => journal.put(processorId, List(repr))
24-
case Some(list) => journal.put(processorId, repr :: list)
25-
}
17+
override def asyncWriteMessages(messages: Seq[PersistentRepr]): Future[Unit] = Future {
18+
log.debug("writeMessages for {} persistent messages", messages.size)
19+
messages.foreach { repr =>
20+
import repr._
21+
val journalForPersistenceId = journal.get(persistenceId)
22+
if (journalForPersistenceId.isEmpty) {
23+
journal.put(processorId, List(repr))
24+
} else {
25+
journalForPersistenceId foreach (xs => journal.put(processorId, repr :: xs))
2626
}
2727
}
2828
}
2929

30-
override def asyncDeleteMessagesTo(persistenceId: String, toSequenceNr: Long, permanent: Boolean): Future[Unit] = {
31-
Future[Unit] {
32-
val perm = permanent
33-
val pid = persistenceId
34-
val toSeq = toSequenceNr
35-
log.debug("asyncDeleteMessagesTo for processorId: {} to sequenceNr: {}, permanent: {}", pid, toSeq, perm)
36-
perm match {
37-
case true =>
38-
journal.get(pid) match {
39-
case None =>
40-
case Some(list) => journal.put(pid, list.filterNot(_.sequenceNr <= toSeq))
41-
}
42-
case false =>
43-
journal.get(pid) match {
44-
case None =>
45-
case Some(list) => journal.put(pid, list.map { repr =>
46-
if(repr.sequenceNr <= toSeq) repr.update(deleted = true) else repr
47-
})
48-
}
30+
override def asyncDeleteMessagesTo(persistenceId: String, toSequenceNr: Long, permanent: Boolean): Future[Unit] = Future {
31+
log.debug("asyncDeleteMessagesTo for processorId: {} to sequenceNr: {}, permanent: {}", persistenceId, toSequenceNr, permanent)
32+
if (permanent) {
33+
journal.get(persistenceId) foreach { list =>
34+
journal.put(persistenceId, list.filterNot(_.sequenceNr <= toSequenceNr))
35+
}
36+
} else {
37+
journal.get(persistenceId) foreach { list =>
38+
journal.put(persistenceId, list.map { repr => if (repr.sequenceNr <= toSequenceNr) repr.update(deleted = true) else repr })
4939
}
5040
}
5141
}
5242

5343
@scala.deprecated("writeConfirmations will be removed, since Channels will be removed.")
54-
override def asyncWriteConfirmations(confirmations: Seq[PersistentConfirmation]): Future[Unit] = {
55-
Future[Unit] {
56-
val confirms = confirmations
57-
log.debug("writeConfirmations for {} messages", confirms.size)
58-
confirms.foreach { confirmation =>
59-
import confirmation._
60-
journal.get(persistenceId) match {
61-
case None =>
62-
case Some(list) =>
63-
journal.put(persistenceId, list.map { msg =>
64-
if(msg.sequenceNr == sequenceNr) {
65-
val confirmationIds = msg.confirms :+ confirmation.channelId
66-
msg.update(confirms = confirmationIds)
67-
} else msg
68-
})
69-
}
44+
override def asyncWriteConfirmations(confirmations: Seq[PersistentConfirmation]): Future[Unit] = Future {
45+
log.debug("writeConfirmations for {} messages", confirmations.size)
46+
confirmations.foreach { confirmation =>
47+
import confirmation._
48+
journal.get(persistenceId).foreach { list =>
49+
journal.put(persistenceId, list.map { msg =>
50+
if (msg.sequenceNr == sequenceNr) {
51+
val confirmationIds = msg.confirms :+ confirmation.channelId
52+
msg.update(confirms = confirmationIds)
53+
} else msg
54+
})
7055
}
7156
}
7257
}
7358

7459
@scala.deprecated("asyncDeleteMessages will be removed.")
75-
override def asyncDeleteMessages(messageIds: Seq[PersistentId], permanent: Boolean): Future[Unit] = {
76-
Future[Unit] {
77-
val mids = messageIds
78-
val perm = permanent
79-
log.debug("Async delete {} messages, permanent: {}", mids.size, perm)
80-
81-
mids.foreach { persistentId =>
82-
import persistentId._
83-
perm match {
84-
case true =>
85-
journal.get(processorId) match {
86-
case None =>
87-
case Some(list) => journal.put(processorId, list.filterNot(_.sequenceNr == sequenceNr))
88-
}
89-
case false =>
90-
journal.get(processorId) match {
91-
case None =>
92-
case Some(list) => journal.put(processorId, list.map { repr =>
93-
if(repr.sequenceNr == sequenceNr) repr.update(deleted = true) else repr
94-
})
95-
}
60+
override def asyncDeleteMessages(messageIds: Seq[PersistentId], permanent: Boolean): Future[Unit] = Future {
61+
log.debug("Async delete {} messages, permanent: {}", messageIds.size, permanent)
62+
messageIds.foreach { persistentId =>
63+
import persistentId._
64+
if (permanent) {
65+
journal.get(processorId) foreach { list =>
66+
journal.put(processorId, list.filterNot(_.sequenceNr == sequenceNr))
67+
}
68+
} else {
69+
journal.get(processorId) foreach { list =>
70+
journal.put(processorId, list.map { repr =>
71+
if (repr.sequenceNr == sequenceNr) repr.update(deleted = true) else repr
72+
})
9673
}
9774
}
9875
}
9976
}
10077

101-
override def asyncReadHighestSequenceNr(persistenceId: String, fromSequenceNr: Long): Future[Long] = {
102-
Future[Long] {
103-
val pid = persistenceId
104-
val fromSeq = fromSequenceNr
105-
log.debug("Async read for highest sequence number for processorId: {} (hint, seek from nr: {})", pid, fromSeq)
106-
journal.get(pid) match {
107-
case None | Some(Nil) => 0
108-
case Some(list) => list.map(_.sequenceNr).max
109-
}
78+
override def asyncReadHighestSequenceNr(persistenceId: String, fromSequenceNr: Long): Future[Long] = Future {
79+
log.debug("Async read for highest sequence number for processorId: {} (hint, seek from nr: {})", persistenceId, fromSequenceNr)
80+
journal.get(persistenceId) match {
81+
case None | Some(Nil) => 0
82+
case Some(list) => list.map(_.sequenceNr).max
11083
}
11184
}
11285

113-
override def asyncReplayMessages(persistenceId: String, fromSequenceNr: Long, toSequenceNr: Long, max: Long)(replayCallback: (PersistentRepr) => Unit): Future[Unit] = {
114-
Future[Unit] {
115-
val pid = persistenceId
116-
val fromSeq = fromSequenceNr
117-
val toSeq = toSequenceNr
118-
val limit = max
119-
val replay = replayCallback
120-
121-
log.debug("Async replay for processorId {}, from sequenceNr: {}, to sequenceNr: {} with max records: {}", pid, fromSeq, toSeq, limit)
122-
123-
journal.get(pid) match {
124-
case None =>
125-
case Some(list) =>
126-
val takeMax = if(limit >= java.lang.Integer.MAX_VALUE) java.lang.Integer.MAX_VALUE else limit.toInt
127-
list.filter { repr =>
128-
repr.sequenceNr >= fromSeq && repr.sequenceNr <= toSeq
129-
}.sortBy(_.sequenceNr)
130-
.take(takeMax).foreach(replay)
131-
}
86+
override def asyncReplayMessages(persistenceId: String, fromSequenceNr: Long, toSequenceNr: Long, max: Long)(replayCallback: (PersistentRepr) => Unit): Future[Unit] = Future {
87+
log.debug("Async replay for processorId {}, from sequenceNr: {}, to sequenceNr: {} with max records: {}", persistenceId, fromSequenceNr, toSequenceNr, max)
88+
journal.get(persistenceId) foreach { list =>
89+
val takeMax = if (max >= java.lang.Integer.MAX_VALUE) java.lang.Integer.MAX_VALUE else max.toInt
90+
list.filter { repr =>
91+
repr.sequenceNr >= fromSequenceNr && repr.sequenceNr <= toSequenceNr
92+
}.sortBy(_.sequenceNr)
93+
.take(takeMax).foreach(replayCallback)
13294
}
13395
}
13496
}

‎src/main/scala/akka/persistence/inmemory/snapshot/InMemorySnapshotStore.scala

+20-33
Original file line numberDiff line numberDiff line change
@@ -1,53 +1,40 @@
11
package akka.persistence.inmemory.snapshot
22

3+
import java.util.concurrent.ConcurrentHashMap
4+
35
import akka.actor.ActorLogging
4-
import akka.persistence.serialization.Snapshot
56
import akka.persistence.snapshot.SnapshotStore
6-
import akka.persistence.{SnapshotMetadata, Persistence, SelectedSnapshot, SnapshotSelectionCriteria}
7-
import akka.serialization.SerializationExtension
8-
import java.util.concurrent.ConcurrentHashMap
7+
import akka.persistence.{SelectedSnapshot, SnapshotMetadata, SnapshotSelectionCriteria}
8+
99
import scala.collection.JavaConverters._
1010
import scala.concurrent.Future
1111

1212
class InMemorySnapshotStore extends SnapshotStore with ActorLogging {
13-
implicit val system = context.system
14-
val extension = Persistence(context.system)
15-
val serialization = SerializationExtension(context.system)
16-
implicit val executionContext = context.system.dispatcher
13+
implicit val ec = context.system.dispatcher
1714

1815
val snapshots: scala.collection.mutable.Map[SnapshotMetadata, Any] = new ConcurrentHashMap[SnapshotMetadata, Any].asScala
1916

20-
override def loadAsync(persistenceId: String, criteria: SnapshotSelectionCriteria): Future[Option[SelectedSnapshot]] = {
21-
Future[Option[SelectedSnapshot]] {
22-
val pid = persistenceId
23-
val crit = criteria
24-
log.debug("loading for persistenceId: {}, criteria: {}", pid, crit)
25-
val snapshotEntries = snapshots.keySet.toList.filter { meta =>
26-
meta.persistenceId == persistenceId && meta.sequenceNr <= criteria.maxSequenceNr
27-
}.filterNot(_.timestamp > criteria.maxTimestamp)
28-
.sortBy(_.sequenceNr)
29-
.reverse.headOption
30-
31-
snapshotEntries match {
32-
case None => None
33-
case Some(meta) => snapshots.get(meta) match {
34-
case None => None
35-
case Some(value) => Some(SelectedSnapshot(meta, value))
36-
}
17+
override def loadAsync(persistenceId: String, criteria: SnapshotSelectionCriteria): Future[Option[SelectedSnapshot]] = Future {
18+
log.debug("loading for persistenceId: {}, criteria: {}", persistenceId, criteria)
19+
snapshots.keySet.toList.filter { meta =>
20+
meta.persistenceId == persistenceId && meta.sequenceNr <= criteria.maxSequenceNr
21+
}.filterNot(_.timestamp > criteria.maxTimestamp)
22+
.sortBy(_.sequenceNr)
23+
.reverse.headOption
24+
.flatMap { (meta: SnapshotMetadata) =>
25+
snapshots.get(meta) map { (snapshot: Any) =>
26+
SelectedSnapshot(meta, snapshot)
3727
}
3828
}
3929
}
4030

41-
override def saveAsync(metadata: SnapshotMetadata, snapshot: Any): Future[Unit] = {
42-
Future[Unit] {
43-
val meta = metadata
44-
val snap = snapshot
45-
log.debug("Saving metadata: {}, snapshot: {}", meta, snap)
46-
snapshots.put(meta, snap)
47-
}
31+
override def saveAsync(metadata: SnapshotMetadata, snapshot: Any): Future[Unit] = Future {
32+
log.debug("Saving metadata: {}, snapshot: {}", metadata, snapshot)
33+
snapshots.put(metadata, snapshot)
4834
}
4935

50-
override def saved(metadata: SnapshotMetadata): Unit = log.debug("Saved: {}", metadata)
36+
override def saved(metadata: SnapshotMetadata): Unit =
37+
log.debug("Saved: {}", metadata)
5138

5239
override def delete(metadata: SnapshotMetadata): Unit = {
5340
log.debug("Deleting: {}", metadata)

0 commit comments

Comments
 (0)