=per #3774 Persistence activator templates
This commit is contained in:
Martin Krasser
Patrik Nordwall
parent
1e86e6436b
commit
cf6e61e45a
31 changed files with 518 additions and 76 deletions
|
|
@ -0,0 +1,62 @@
|
|||
/**
|
||||
* Copyright (C) 2009-2014 Typesafe Inc. <http://www.typesafe.com>
|
||||
*/
|
||||
|
||||
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);
|
||||
}
|
||||
}
|
||||
|
|
@ -0,0 +1,141 @@
|
|||
/**
|
||||
* Copyright (C) 2009-2014 Typesafe Inc. <http://www.typesafe.com>
|
||||
*/
|
||||
|
||||
package sample.persistence;
|
||||
|
||||
//#eventsourced-example
|
||||
import java.io.Serializable;
|
||||
import java.util.ArrayList;
|
||||
|
||||
import akka.actor.*;
|
||||
import akka.japi.Procedure;
|
||||
import akka.persistence.*;
|
||||
|
||||
import static java.util.Arrays.asList;
|
||||
|
||||
class Cmd implements Serializable {
|
||||
private final String data;
|
||||
|
||||
public Cmd(String data) {
|
||||
this.data = data;
|
||||
}
|
||||
|
||||
public String getData() {
|
||||
return data;
|
||||
}
|
||||
}
|
||||
|
||||
class Evt implements Serializable {
|
||||
private final String data;
|
||||
|
||||
public Evt(String data) {
|
||||
this.data = data;
|
||||
}
|
||||
|
||||
public String getData() {
|
||||
return data;
|
||||
}
|
||||
}
|
||||
|
||||
class ExampleState implements Serializable {
|
||||
private final ArrayList<String> events;
|
||||
|
||||
public ExampleState() {
|
||||
this(new ArrayList<String>());
|
||||
}
|
||||
|
||||
public ExampleState(ArrayList<String> events) {
|
||||
this.events = events;
|
||||
}
|
||||
|
||||
public ExampleState copy() {
|
||||
return new ExampleState(new ArrayList<String>(events));
|
||||
}
|
||||
|
||||
public void update(Evt evt) {
|
||||
events.add(evt.getData());
|
||||
}
|
||||
|
||||
public int size() {
|
||||
return events.size();
|
||||
}
|
||||
|
||||
@Override
|
||||
public String toString() {
|
||||
return events.toString();
|
||||
}
|
||||
}
|
||||
|
||||
class ExampleProcessor extends UntypedEventsourcedProcessor {
|
||||
private ExampleState state = new ExampleState();
|
||||
|
||||
public int getNumEvents() {
|
||||
return state.size();
|
||||
}
|
||||
|
||||
public void onReceiveRecover(Object msg) {
|
||||
if (msg instanceof Evt) {
|
||||
state.update((Evt) msg);
|
||||
} else if (msg instanceof SnapshotOffer) {
|
||||
state = (ExampleState)((SnapshotOffer)msg).snapshot();
|
||||
}
|
||||
}
|
||||
|
||||
public void onReceiveCommand(Object msg) {
|
||||
if (msg instanceof Cmd) {
|
||||
final String data = ((Cmd)msg).getData();
|
||||
final Evt evt1 = new Evt(data + "-" + getNumEvents());
|
||||
final Evt evt2 = new Evt(data + "-" + (getNumEvents() + 1));
|
||||
persist(asList(evt1, evt2), new Procedure<Evt>() {
|
||||
public void apply(Evt evt) throws Exception {
|
||||
state.update(evt);
|
||||
if (evt.equals(evt2)) {
|
||||
getContext().system().eventStream().publish(evt);
|
||||
if (data.equals("foo")) getContext().become(otherCommandHandler);
|
||||
}
|
||||
}
|
||||
});
|
||||
} else if (msg.equals("snap")) {
|
||||
// IMPORTANT: create a copy of snapshot
|
||||
// because ExampleState is mutable !!!
|
||||
saveSnapshot(state.copy());
|
||||
} else if (msg.equals("print")) {
|
||||
System.out.println(state);
|
||||
}
|
||||
}
|
||||
|
||||
private Procedure<Object> otherCommandHandler = new Procedure<Object>() {
|
||||
public void apply(Object msg) throws Exception {
|
||||
if (msg instanceof Cmd && ((Cmd)msg).getData().equals("bar")) {
|
||||
persist(new Evt("bar-" + getNumEvents()), new Procedure<Evt>() {
|
||||
public void apply(Evt event) throws Exception {
|
||||
state.update(event);
|
||||
getContext().unbecome();
|
||||
}
|
||||
});
|
||||
unstashAll();
|
||||
} else {
|
||||
stash();
|
||||
}
|
||||
}
|
||||
};
|
||||
}
|
||||
//#eventsourced-example
|
||||
|
||||
public class EventsourcedExample {
|
||||
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");
|
||||
|
||||
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);
|
||||
|
||||
Thread.sleep(1000);
|
||||
system.shutdown();
|
||||
}
|
||||
}
|
||||
|
|
@ -0,0 +1,55 @@
|
|||
/**
|
||||
* Copyright (C) 2009-2014 Typesafe Inc. <http://www.typesafe.com>
|
||||
*/
|
||||
|
||||
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();
|
||||
}
|
||||
}
|
||||
|
|
@ -0,0 +1,75 @@
|
|||
/**
|
||||
* Copyright (C) 2009-2014 Typesafe Inc. <http://www.typesafe.com>
|
||||
*/
|
||||
|
||||
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();
|
||||
}
|
||||
}
|
||||
|
|
@ -0,0 +1,75 @@
|
|||
/**
|
||||
* Copyright (C) 2009-2014 Typesafe Inc. <http://www.typesafe.com>
|
||||
*/
|
||||
|
||||
package sample.persistence;
|
||||
|
||||
import java.io.Serializable;
|
||||
import java.util.ArrayList;
|
||||
|
||||
import akka.actor.*;
|
||||
import akka.persistence.*;
|
||||
|
||||
public class SnapshotExample {
|
||||
public static class ExampleState implements Serializable {
|
||||
private final ArrayList<String> received;
|
||||
|
||||
public ExampleState() {
|
||||
this(new ArrayList<String>());
|
||||
}
|
||||
|
||||
public ExampleState(ArrayList<String> received) {
|
||||
this.received = received;
|
||||
}
|
||||
|
||||
public ExampleState copy() {
|
||||
return new ExampleState(new ArrayList<String>(received));
|
||||
}
|
||||
|
||||
public void update(String s) {
|
||||
received.add(s);
|
||||
}
|
||||
|
||||
@Override
|
||||
public String toString() {
|
||||
return received.toString();
|
||||
}
|
||||
}
|
||||
|
||||
public static class ExampleProcessor extends UntypedProcessor {
|
||||
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")) {
|
||||
System.out.println("current state = " + state);
|
||||
} else if (message.equals("snap")) {
|
||||
// IMPORTANT: create a copy of snapshot
|
||||
// because ExampleState is mutable !!!
|
||||
saveSnapshot(state.copy());
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
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");
|
||||
|
||||
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);
|
||||
|
||||
Thread.sleep(1000);
|
||||
system.shutdown();
|
||||
}
|
||||
}
|
||||
|
|
@ -0,0 +1,82 @@
|
|||
/**
|
||||
* Copyright (C) 2009-2014 Typesafe Inc. <http://www.typesafe.com>
|
||||
*/
|
||||
|
||||
package sample.persistence;
|
||||
|
||||
import java.util.concurrent.TimeUnit;
|
||||
|
||||
import scala.concurrent.duration.Duration;
|
||||
|
||||
import akka.actor.*;
|
||||
import akka.persistence.*;
|
||||
|
||||
public class ViewExample {
|
||||
public static class ExampleProcessor extends UntypedProcessor {
|
||||
@Override
|
||||
public String processorId() {
|
||||
return "processor-5";
|
||||
}
|
||||
|
||||
@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 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;
|
||||
|
||||
@Override
|
||||
public String viewId() {
|
||||
return "view-5";
|
||||
}
|
||||
|
||||
@Override
|
||||
public String processorId() {
|
||||
return "processor-5";
|
||||
}
|
||||
|
||||
@Override
|
||||
public void onReceive(Object message) throws Exception {
|
||||
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());
|
||||
} else if (message instanceof SnapshotOffer) {
|
||||
SnapshotOffer so = (SnapshotOffer)message;
|
||||
numReplicated = (Integer)so.snapshot();
|
||||
System.out.println(String.format("view received snapshot offer %s (metadata = %s)", numReplicated, so.metadata()));
|
||||
} else if (message.equals("snap")) {
|
||||
saveSnapshot(numReplicated);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
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 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(5, TimeUnit.SECONDS), view, "snap", system.dispatcher(), null);
|
||||
}
|
||||
}
|
||||
Loading…
Add table
Add a link
Reference in a new issue