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

=per #15429 Rewrite persistence documentation and samples for 2.3.4 changes

Browse source 
(cherry picked from commit 02351e32f110a8c4a249f0f3f84bae5898d1a836)

Conflicts:
	akka-samples/akka-sample-persistence-java-lambda/tutorial/index.html
	akka-samples/akka-sample-persistence-java/tutorial/index.html
	akka-samples/akka-sample-persistence-scala/build.sbt
	akka-samples/akka-sample-persistence-scala/src/main/scala/sample/persistence/ConversationRecoveryExample.scala
	akka-samples/akka-sample-persistence-scala/src/main/scala/sample/persistence/PersistentActorExample.scala
	akka-samples/akka-sample-persistence-scala/src/main/scala/sample/persistence/ProcessorChannelExample.scala
	akka-samples/akka-sample-persistence-scala/src/main/scala/sample/persistence/ProcessorChannelRemoteExample.scala
	akka-samples/akka-sample-persistence-scala/src/main/scala/sample/persistence/SnapshotExample.scala
	akka-samples/akka-sample-persistence-scala/src/main/scala/sample/persistence/StreamExample.scala
	akka-samples/akka-sample-persistence-scala/tutorial/index.html
This commit is contained in:
Patrik Nordwall 2014-06-25 12:51:21 +02:00
parent 062d304b73
commit d6ffdf521c
35 changed files with 1091 additions and 2276 deletions
Download patch file Download diff file Expand all files Collapse all files

58
akka-samples/akka-sample-persistence-java/src/main/java/sample/persistence/ConversationRecoveryExample.java
View file

@ -1,58 +0,0 @@
package sample.persistence;
import akka.actor.*;
import akka.persistence.*;
public class ConversationRecoveryExample {
public static String PING = "PING";
public static String PONG = "PONG";
public static class Ping extends UntypedProcessor {
private final ActorRef pongChannel = getContext().actorOf(Channel.props(), "pongChannel");
private int counter = 0;
@Override
public void onReceive(Object message) throws Exception {
if (message instanceof ConfirmablePersistent) {
ConfirmablePersistent msg = (ConfirmablePersistent)message;
if (msg.payload().equals(PING)) {
counter += 1;
System.out.println(String.format("received ping %d times", counter));
msg.confirm();
if (!recoveryRunning()) Thread.sleep(1000);
pongChannel.tell(Deliver.create(msg.withPayload(PONG), getSender().path()), getSelf());
}
} else if (message.equals("init") && counter == 0) {
pongChannel.tell(Deliver.create(Persistent.create(PONG), getSender().path()), getSelf());
}
}
}
public static class Pong extends UntypedProcessor {
private final ActorRef pingChannel = getContext().actorOf(Channel.props(), "pingChannel");
private int counter = 0;
@Override
public void onReceive(Object message) throws Exception {
if (message instanceof ConfirmablePersistent) {
ConfirmablePersistent msg = (ConfirmablePersistent)message;
if (msg.payload().equals(PONG)) {
counter += 1;
System.out.println(String.format("received pong %d times", counter));
msg.confirm();
if (!recoveryRunning()) Thread.sleep(1000);
pingChannel.tell(Deliver.create(msg.withPayload(PING), getSender().path()), getSelf());
}
}
}
}
public static void main(String... args) throws Exception {
final ActorSystem system = ActorSystem.create("example");
final ActorRef ping = system.actorOf(Props.create(Ping.class), "ping");
final ActorRef pong = system.actorOf(Props.create(Pong.class), "pong");
ping.tell("init", pong);
}
}

24
akka-samples/akka-sample-persistence-java/src/main/java/sample/persistence/PersistentActorExample.java
View file

@ -10,7 +10,6 @@ import akka.persistence.UntypedPersistentActor;
import java.io.Serializable;
import java.util.ArrayList;
import static java.util.Arrays.asList;
class Cmd implements Serializable {
@ -66,21 +65,25 @@ class ExampleState implements Serializable {
}
}
class ExampleProcessor extends UntypedPersistentActor {
class ExamplePersistentActor extends UntypedPersistentActor {
private ExampleState state = new ExampleState();
public int getNumEvents() {
return state.size();
}
@Override
public void onReceiveRecover(Object msg) {
if (msg instanceof Evt) {
state.update((Evt) msg);
} else if (msg instanceof SnapshotOffer) {
state = (ExampleState)((SnapshotOffer)msg).snapshot();
} else {
unhandled(msg);
}
}
@Override
public void onReceiveCommand(Object msg) {
if (msg instanceof Cmd) {
final String data = ((Cmd)msg).getData();
@ -100,6 +103,8 @@ class ExampleProcessor extends UntypedPersistentActor {
saveSnapshot(state.copy());
} else if (msg.equals("print")) {
System.out.println(state);
} else {
unhandled(msg);
}
}
}
@ -108,14 +113,15 @@ class ExampleProcessor extends UntypedPersistentActor {
public class PersistentActorExample {
public static void main(String... args) throws Exception {
final ActorSystem system = ActorSystem.create("example");
final ActorRef processor = system.actorOf(Props.create(ExampleProcessor.class), "processor-4-java");
final ActorRef persistentActor =
system.actorOf(Props.create(ExamplePersistentActor.class), "persistentActor-4-java");
processor.tell(new Cmd("foo"), null);
processor.tell(new Cmd("baz"), null);
processor.tell(new Cmd("bar"), null);
processor.tell("snap", null);
processor.tell(new Cmd("buzz"), null);
processor.tell("print", null);
persistentActor.tell(new Cmd("foo"), null);
persistentActor.tell(new Cmd("baz"), null);
persistentActor.tell(new Cmd("bar"), null);
persistentActor.tell("snap", null);
persistentActor.tell(new Cmd("buzz"), null);
persistentActor.tell("print", null);
Thread.sleep(1000);
system.shutdown();

71
akka-samples/akka-sample-persistence-java/src/main/java/sample/persistence/PersistentActorFailureExample.java Normal file
View file

@ -0,0 +1,71 @@
package sample.persistence;
import java.util.ArrayList;
import akka.japi.Procedure;
import akka.actor.*;
import akka.persistence.*;
public class PersistentActorFailureExample {
public static class ExamplePersistentActor extends UntypedPersistentActor {
private ArrayList<Object> received = new ArrayList<Object>();
@Override
public void onReceiveCommand(Object message) throws Exception {
if (message.equals("boom")) {
throw new Exception("boom");
} else if (message.equals("print")) {
System.out.println("received " + received);
} else if (message instanceof String) {
String s = (String) message;
persist(s, new Procedure<String>() {
public void apply(String evt) throws Exception {
received.add(evt);
}
});
} else {
unhandled(message);
}
}
@Override
public void onReceiveRecover(Object message) {
if (message instanceof String) {
received.add((String) message);
} else {
unhandled(message);
}
}
}
public static void main(String... args) throws Exception {
final ActorSystem system = ActorSystem.create("example");
final ActorRef persistentActor = system.actorOf(Props.create(ExamplePersistentActor.class), "persistentActor-2");
persistentActor.tell("a", null);
persistentActor.tell("print", null);
persistentActor.tell("boom", null);
persistentActor.tell("print", null);
persistentActor.tell("b", null);
persistentActor.tell("print", null);
persistentActor.tell("c", null);
persistentActor.tell("print", null);
// Will print in a first run (i.e. with empty journal):
// received [a]
// received [a, b]
// received [a, b, c]
// Will print in a second run:
// received [a, b, c, a]
// received [a, b, c, a, b]
// received [a, b, c, a, b, c]
// etc ...
Thread.sleep(1000);
system.shutdown();
}
}

51
akka-samples/akka-sample-persistence-java/src/main/java/sample/persistence/ProcessorChannelExample.java
View file

@ -1,51 +0,0 @@
package sample.persistence;
import akka.actor.*;
import akka.persistence.*;
public class ProcessorChannelExample {
public static class ExampleProcessor extends UntypedProcessor {
private ActorRef destination;
private ActorRef channel;
public ExampleProcessor(ActorRef destination) {
this.destination = destination;
this.channel = getContext().actorOf(Channel.props(), "channel");
}
@Override
public void onReceive(Object message) throws Exception {
if (message instanceof Persistent) {
Persistent msg = (Persistent)message;
System.out.println("processed " + msg.payload());
channel.tell(Deliver.create(msg.withPayload("processed " + msg.payload()), destination.path()), getSelf());
} else if (message instanceof String) {
System.out.println("reply = " + message);
}
}
}
public static class ExampleDestination extends UntypedActor {
@Override
public void onReceive(Object message) throws Exception {
if (message instanceof ConfirmablePersistent) {
ConfirmablePersistent msg = (ConfirmablePersistent)message;
System.out.println("received " + msg.payload());
getSender().tell(String.format("re: %s (%d)", msg.payload(), msg.sequenceNr()), null);
msg.confirm();
}
}
}
public static void main(String... args) throws Exception {
final ActorSystem system = ActorSystem.create("example");
final ActorRef destination = system.actorOf(Props.create(ExampleDestination.class));
final ActorRef processor = system.actorOf(Props.create(ExampleProcessor.class, destination), "processor-1");
processor.tell(Persistent.create("a"), null);
processor.tell(Persistent.create("b"), null);
Thread.sleep(1000);
system.shutdown();
}
}

71
akka-samples/akka-sample-persistence-java/src/main/java/sample/persistence/ProcessorFailureExample.java
View file

@ -1,71 +0,0 @@
package sample.persistence;
import java.util.ArrayList;
import scala.Option;
import akka.actor.*;
import akka.persistence.*;
public class ProcessorFailureExample {
public static class ExampleProcessor extends UntypedProcessor {
private ArrayList<Object> received = new ArrayList<Object>();
@Override
public void onReceive(Object message) throws Exception {
if (message instanceof Persistent) {
Persistent persistent = (Persistent)message;
if (persistent.payload() == "boom") {
throw new Exception("boom");
} else {
received.add(persistent.payload());
}
} else if (message == "boom") {
throw new Exception("boom");
} else if (message == "print") {
System.out.println("received " + received);
}
}
@Override
public void preRestart(Throwable reason, Option<Object> message) {
if (message.isDefined() && message.get() instanceof Persistent) {
deleteMessage(((Persistent) message.get()).sequenceNr(), false);
}
super.preRestart(reason, message);
}
}
public static void main(String... args) throws Exception {
final ActorSystem system = ActorSystem.create("example");
final ActorRef processor = system.actorOf(Props.create(ExampleProcessor.class), "processor-2");
processor.tell(Persistent.create("a"), null);
processor.tell("print", null);
processor.tell("boom", null);
processor.tell("print", null);
processor.tell(Persistent.create("b"), null);
processor.tell("print", null);
processor.tell(Persistent.create("boom"), null);
processor.tell("print", null);
processor.tell(Persistent.create("c"), null);
processor.tell("print", null);
// Will print in a first run (i.e. with empty journal):
// received [a]
// received [a, b]
// received [a, b, c]
// Will print in a second run:
// received [a, b, c, a]
// received [a, b, c, a, b]
// received [a, b, c, a, b, c]
// etc ...
Thread.sleep(1000);
system.shutdown();
}
}

54
akka-samples/akka-sample-persistence-java/src/main/java/sample/persistence/SnapshotExample.java
View file

@ -5,6 +5,7 @@ import java.util.ArrayList;
import akka.actor.*;
import akka.persistence.*;
import akka.japi.Procedure;
public class SnapshotExample {
public static class ExampleState implements Serializable {
@ -32,38 +33,57 @@ public class SnapshotExample {
}
}
public static class ExampleProcessor extends UntypedProcessor {
public static class ExamplePersistentActor extends UntypedPersistentActor {
private ExampleState state = new ExampleState();
@Override
public void onReceive(Object message) throws Exception {
if (message instanceof Persistent) {
Persistent persistent = (Persistent)message;
state.update(String.format("%s-%d", persistent.payload(), persistent.sequenceNr()));
} else if (message instanceof SnapshotOffer) {
ExampleState s = (ExampleState)((SnapshotOffer)message).snapshot();
System.out.println("offered state = " + s);
state = s;
} else if (message.equals("print")) {
public void onReceiveCommand(Object message) {
if (message.equals("print")) {
System.out.println("current state = " + state);
} else if (message.equals("snap")) {
// IMPORTANT: create a copy of snapshot
// because ExampleState is mutable !!!
saveSnapshot(state.copy());
} else if (message instanceof SaveSnapshotSuccess) {
// ...
} else if (message instanceof SaveSnapshotFailure) {
// ...
} else if (message instanceof String) {
String s = (String) message;
persist(s, new Procedure<String>() {
public void apply(String evt) throws Exception {
state.update(evt);
}
});
} else {
unhandled(message);
}
}
@Override
public void onReceiveRecover(Object message) {
if (message instanceof SnapshotOffer) {
ExampleState s = (ExampleState)((SnapshotOffer)message).snapshot();
System.out.println("offered state = " + s);
state = s;
} else if (message instanceof String) {
state.update((String) message);
} else {
unhandled(message);
}
}
}
public static void main(String... args) throws Exception {
final ActorSystem system = ActorSystem.create("example");
final ActorRef processor = system.actorOf(Props.create(ExampleProcessor.class), "processor-3-java");
final ActorRef persistentActor = system.actorOf(Props.create(ExamplePersistentActor.class), "persistentActor-3-java");
processor.tell(Persistent.create("a"), null);
processor.tell(Persistent.create("b"), null);
processor.tell("snap", null);
processor.tell(Persistent.create("c"), null);
processor.tell(Persistent.create("d"), null);
processor.tell("print", null);
persistentActor.tell("a", null);
persistentActor.tell("b", null);
persistentActor.tell("snap", null);
persistentActor.tell("c", null);
persistentActor.tell("d", null);
persistentActor.tell("print", null);
Thread.sleep(1000);
system.shutdown();

58
akka-samples/akka-sample-persistence-java/src/main/java/sample/persistence/ViewExample.java
View file

@ -1,31 +1,47 @@
package sample.persistence;
import java.util.concurrent.TimeUnit;
import scala.concurrent.duration.Duration;
import akka.actor.*;
import akka.persistence.*;
import akka.japi.Procedure;
public class ViewExample {
public static class ExampleProcessor extends UntypedProcessor {
public static class ExamplePersistentActor extends UntypedPersistentActor {
private int count = 1;
@Override
public String processorId() {
return "processor-5";
public String persistenceId() {
return "persistentActor-5";
}
@Override
public void onReceiveRecover(Object message) {
if (message instanceof String) {
count += 1;
} else {
unhandled(message);
}
}
@Override
public void onReceive(Object message) throws Exception {
if (message instanceof Persistent) {
Persistent p = (Persistent)message;
System.out.println(String.format("processor received %s (sequence nr = %d)", p.payload(), p.sequenceNr()));
}
public void onReceiveCommand(Object message) {
if (message instanceof String) {
String s = (String) message;
System.out.println(String.format("persistentActor received %s (nr = %d)", s, count));
persist(s + count, new Procedure<String>() {
public void apply(String evt) {
count += 1;
}
});
} else {
unhandled(message);
}
}
}
public static class ExampleView extends UntypedView {
private final ActorRef destination = getContext().actorOf(Props.create(ExampleDestination.class));
private final ActorRef channel = getContext().actorOf(Channel.props("channel"));
private int numReplicated = 0;
@ -36,7 +52,7 @@ public class ViewExample {
@Override
public String persistenceId() {
return "processor-5";
return "persistentActor-5";
}
@Override
@ -44,8 +60,7 @@ public class ViewExample {
if (message instanceof Persistent) {
Persistent p = (Persistent)message;
numReplicated += 1;
System.out.println(String.format("view received %s (sequence nr = %d, num replicated = %d)", p.payload(), p.sequenceNr(), numReplicated));
channel.tell(Deliver.create(p.withPayload("replicated-" + p.payload()), destination.path()), getSelf());
System.out.println(String.format("view received %s (num replicated = %d)", p.payload(), numReplicated));
} else if (message instanceof SnapshotOffer) {
SnapshotOffer so = (SnapshotOffer)message;
numReplicated = (Integer)so.snapshot();
@ -56,23 +71,12 @@ public class ViewExample {
}
}
public static class ExampleDestination extends UntypedActor {
@Override
public void onReceive(Object message) throws Exception {
if (message instanceof ConfirmablePersistent) {
ConfirmablePersistent cp = (ConfirmablePersistent)message;
System.out.println(String.format("destination received %s (sequence nr = %s)", cp.payload(), cp.sequenceNr()));
cp.confirm();
}
}
}
public static void main(String... args) throws Exception {
final ActorSystem system = ActorSystem.create("example");
final ActorRef processor = system.actorOf(Props.create(ExampleProcessor.class));
final ActorRef persistentActor = system.actorOf(Props.create(ExamplePersistentActor.class));
final ActorRef view = system.actorOf(Props.create(ExampleView.class));
system.scheduler().schedule(Duration.Zero(), Duration.create(2, TimeUnit.SECONDS), processor, Persistent.create("scheduled"), system.dispatcher(), null);
system.scheduler().schedule(Duration.Zero(), Duration.create(2, TimeUnit.SECONDS), persistentActor, "scheduled", system.dispatcher(), null);
system.scheduler().schedule(Duration.Zero(), Duration.create(5, TimeUnit.SECONDS), view, "snap", system.dispatcher(), null);
}
}