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

!per,doc,sam #17574 #17626 make recovery a method, move lambda samples

Browse source 
+ LambdaDoc samples now in the docs project
= simplified internal state by removing recoveryPending
= recovery is now triggered in around* method, so user is free to use
  preStart freely - recovery works even if one forgets to call super on
  preStart
This commit is contained in:
Konrad Malawski 2015-06-24 19:58:43 +02:00
parent f38af5fd1a
commit 33fbfec222
23 changed files with 218 additions and 342 deletions
Download patch file Download diff file Expand all files Collapse all files

544
akka-docs/rst/java/code/docs/persistence/LambdaPersistenceDocTest.java Normal file
View file

@ -0,0 +1,544 @@
/**
* Copyright (C) 2009-2015 Typesafe Inc. <http://www.typesafe.com>
*/
package docs.persistence;
import akka.actor.AbstractActor;
import akka.actor.ActorPath;
import akka.actor.ActorRef;
import akka.actor.ActorSystem;
import akka.actor.Props;
import akka.japi.Procedure;
import akka.japi.pf.ReceiveBuilder;
import akka.persistence.*;
import scala.Option;
import scala.concurrent.duration.Duration;
import scala.PartialFunction;
import scala.runtime.BoxedUnit;
import java.io.Serializable;
import java.util.concurrent.TimeUnit;
public class LambdaPersistenceDocTest {
public interface SomeOtherMessage {}
public interface PersistentActorMethods {
//#persistence-id
public String persistenceId();
//#persistence-id
//#recovery-status
public boolean recoveryRunning();
public boolean recoveryFinished();
//#recovery-status
}
static Object o2 = new Object() {
abstract class MyPersistentActor1 extends AbstractPersistentActor {
//#recovery-disabled
@Override
public Recovery recovery() {
return Recovery.none();
}
//#recovery-disabled
//#recover-on-restart-disabled
@Override
public void preRestart(Throwable reason, Option<Object> message) {}
//#recover-on-restart-disabled
}
abstract class MyPersistentActor2 extends AbstractPersistentActor {
//#recovery-custom
@Override
public Recovery recovery() {
return Recovery.create(457L);
}
//#recovery-custom
}
class MyPersistentActor4 extends AbstractPersistentActor implements PersistentActorMethods {
//#persistence-id-override
@Override
public String persistenceId() {
return "my-stable-persistence-id";
}
//#persistence-id-override
@Override
public PartialFunction<Object, BoxedUnit> receiveCommand() {
return ReceiveBuilder.
match(String.class, cmd -> {/* ... */}).build();
}
@Override
public PartialFunction<Object, BoxedUnit> receiveRecover() {
return ReceiveBuilder.
match(String.class, evt -> {/* ... */}).build();
}
}
//#recovery-completed
class MyPersistentActor5 extends AbstractPersistentActor {
@Override public String persistenceId() {
return "my-stable-persistence-id";
}
@Override public PartialFunction<Object, BoxedUnit> receiveRecover() {
return ReceiveBuilder.
match(RecoveryCompleted.class, r -> {
// perform init after recovery, before any other messages
// ...
}).
match(String.class, this::handleEvent).build();
}
@Override public PartialFunction<Object, BoxedUnit> receiveCommand() {
return ReceiveBuilder.
match(String.class, s -> s.equals("cmd"),
s -> persist("evt", this::handleEvent)).build();
}
private void handleEvent(String event) {
// update state
// ...
}
}
//#recovery-completed
abstract class MyActor extends AbstractPersistentActor {
//#backoff
@Override
public void preStart() throws Exception {
final Props childProps = Props.create(MyPersistentActor1.class);
final Props props = BackoffSupervisor.props(
childProps,
"myActor",
Duration.create(3, TimeUnit.SECONDS),
Duration.create(30, TimeUnit.SECONDS),
0.2);
context().actorOf(props, "mySupervisor");
super.preStart();
}
//#backoff
}
};
static Object atLeastOnceExample = new Object() {
//#at-least-once-example
class Msg implements Serializable {
private static final long serialVersionUID = 1L;
public final long deliveryId;
public final String s;
public Msg(long deliveryId, String s) {
this.deliveryId = deliveryId;
this.s = s;
}
}
class Confirm implements Serializable {
private static final long serialVersionUID = 1L;
public final long deliveryId;
public Confirm(long deliveryId) {
this.deliveryId = deliveryId;
}
}
class MsgSent implements Serializable {
private static final long serialVersionUID = 1L;
public final String s;
public MsgSent(String s) {
this.s = s;
}
}
class MsgConfirmed implements Serializable {
private static final long serialVersionUID = 1L;
public final long deliveryId;
public MsgConfirmed(long deliveryId) {
this.deliveryId = deliveryId;
}
}
class MyPersistentActor extends AbstractPersistentActorWithAtLeastOnceDelivery {
private final ActorPath destination;
public MyPersistentActor(ActorPath destination) {
this.destination = destination;
}
@Override public String persistenceId() {
return "persistence-id";
}
@Override
public PartialFunction<Object, BoxedUnit> receiveCommand() {
return ReceiveBuilder.
match(String.class, s -> {
persist(new MsgSent(s), evt -> updateState(evt));
}).
match(Confirm.class, confirm -> {
persist(new MsgConfirmed(confirm.deliveryId), evt -> updateState(evt));
}).
build();
}
@Override
public PartialFunction<Object, BoxedUnit> receiveRecover() {
return ReceiveBuilder.
match(Object.class, evt -> updateState(evt)).build();
}
void updateState(Object event) {
if (event instanceof MsgSent) {
final MsgSent evt = (MsgSent) event;
deliver(destination, deliveryId -> new Msg(deliveryId, evt.s));
} else if (event instanceof MsgConfirmed) {
final MsgConfirmed evt = (MsgConfirmed) event;
confirmDelivery(evt.deliveryId);
}
}
}
class MyDestination extends AbstractActor {
public MyDestination() {
receive(ReceiveBuilder.
match(Msg.class, msg -> {
// ...
sender().tell(new Confirm(msg.deliveryId), self());
}).build()
);
}
}
//#at-least-once-example
};
static Object o4 = new Object() {
class MyPersistentActor extends AbstractPersistentActor {
//#save-snapshot
private Object state;
@Override public PartialFunction<Object, BoxedUnit> receiveCommand() {
return ReceiveBuilder.
match(String.class, s -> s.equals("snap"),
s -> saveSnapshot(state)).
match(SaveSnapshotSuccess.class, ss -> {
SnapshotMetadata metadata = ss.metadata();
// ...
}).
match(SaveSnapshotFailure.class, sf -> {
SnapshotMetadata metadata = sf.metadata();
// ...
}).build();
}
//#save-snapshot
@Override public String persistenceId() {
return "persistence-id";
}
@Override public PartialFunction<Object, BoxedUnit> receiveRecover() {
return ReceiveBuilder.
match(RecoveryCompleted.class, r -> {/* ...*/}).build();
}
}
};
static Object o5 = new Object() {
class MyPersistentActor extends AbstractPersistentActor {
//#snapshot-offer
private Object state;
@Override public PartialFunction<Object, BoxedUnit> receiveRecover() {
return ReceiveBuilder.
match(SnapshotOffer.class, s -> {
state = s.snapshot();
// ...
}).
match(String.class, s -> {/* ...*/}).build();
}
//#snapshot-offer
@Override public String persistenceId() {
return "persistence-id";
}
@Override public PartialFunction<Object, BoxedUnit> receiveCommand() {
return ReceiveBuilder.
match(String.class, s -> {/* ...*/}).build();
}
}
class MyActor extends AbstractActor {
ActorRef persistentActor;
public MyActor() {
persistentActor = context().actorOf(Props.create(MyPersistentActor.class));
receive(ReceiveBuilder.
match(Object.class, o -> {/* ... */}).build()
);
}
private void recover() {
//#snapshot-criteria
persistentActor.tell(Recovery.create(
SnapshotSelectionCriteria
.create(457L, System.currentTimeMillis())), null);
//#snapshot-criteria
}
}
};
static Object o9 = new Object() {
//#persist-async
class MyPersistentActor extends AbstractPersistentActor {
@Override public String persistenceId() {
return "my-stable-persistence-id";
}
private void handleCommand(String c) {
sender().tell(c, self());
persistAsync(String.format("evt-%s-1", c), e -> {
sender().tell(e, self());
});
persistAsync(String.format("evt-%s-2", c), e -> {
sender().tell(e, self());
});
}
@Override public PartialFunction<Object, BoxedUnit> receiveRecover() {
return ReceiveBuilder.
match(String.class, this::handleCommand).build();
}
@Override public PartialFunction<Object, BoxedUnit> receiveCommand() {
return ReceiveBuilder.
match(String.class, this::handleCommand).build();
}
}
//#persist-async
public void usage() {
final ActorSystem system = ActorSystem.create("example");
//#persist-async-usage
final ActorRef persistentActor = system.actorOf(Props.create(MyPersistentActor.class));
persistentActor.tell("a", null);
persistentActor.tell("b", null);
// possible order of received messages:
// a
// b
// evt-a-1
// evt-a-2
// evt-b-1
// evt-b-2
//#persist-async-usage
}
};
static Object o10 = new Object() {
//#defer
class MyPersistentActor extends AbstractPersistentActor {
@Override public String persistenceId() {
return "my-stable-persistence-id";
}
private void handleCommand(String c) {
persistAsync(String.format("evt-%s-1", c), e -> {
sender().tell(e, self());
});
persistAsync(String.format("evt-%s-2", c), e -> {
sender().tell(e, self());
});
deferAsync(String.format("evt-%s-3", c), e -> {
sender().tell(e, self());
});
}
@Override public PartialFunction<Object, BoxedUnit> receiveRecover() {
return ReceiveBuilder.
match(String.class, this::handleCommand).build();
}
@Override public PartialFunction<Object, BoxedUnit> receiveCommand() {
return ReceiveBuilder.
match(String.class, this::handleCommand).build();
}
}
//#defer
public void usage() {
final ActorSystem system = ActorSystem.create("example");
final ActorRef sender = null; // your imaginary sender here
//#defer-caller
final ActorRef persistentActor = system.actorOf(Props.create(MyPersistentActor.class));
persistentActor.tell("a", sender);
persistentActor.tell("b", sender);
// order of received messages:
// a
// b
// evt-a-1
// evt-a-2
// evt-a-3
// evt-b-1
// evt-b-2
// evt-b-3
//#defer-caller
}
};
static Object o11 = new Object() {
class MyPersistentActor extends AbstractPersistentActor {
@Override
public String persistenceId() {
return "my-stable-persistence-id";
}
@Override public PartialFunction<Object, BoxedUnit> receiveCommand() {
return ReceiveBuilder.matchAny(event -> {}).build();
}
//#nested-persist-persist
@Override public PartialFunction<Object, BoxedUnit> receiveRecover() {
final Procedure<String> replyToSender = event -> sender().tell(event, self());
return ReceiveBuilder
.match(String.class, msg -> {
persist(String.format("%s-outer-1", msg), event -> {
sender().tell(event, self());
persist(String.format("%s-inner-1", event), replyToSender);
});
persist(String.format("%s-outer-2", msg), event -> {
sender().tell(event, self());
persist(String.format("%s-inner-2", event), replyToSender);
});
})
.build();
}
//#nested-persist-persist
void usage(ActorRef persistentActor) {
//#nested-persist-persist-caller
persistentActor.tell("a", ActorRef.noSender());
persistentActor.tell("b", ActorRef.noSender());
// order of received messages:
// a
// a-outer-1
// a-outer-2
// a-inner-1
// a-inner-2
// and only then process "b"
// b
// b-outer-1
// b-outer-2
// b-inner-1
// b-inner-2
//#nested-persist-persist-caller
}
}
class MyPersistAsyncActor extends AbstractPersistentActor {
@Override
public String persistenceId() {
return "my-stable-persistence-id";
}
@Override public PartialFunction<Object, BoxedUnit> receiveCommand() {
return ReceiveBuilder.matchAny(event -> {}).build();
}
//#nested-persistAsync-persistAsync
@Override public PartialFunction<Object, BoxedUnit> receiveRecover() {
final Procedure<String> replyToSender = event -> sender().tell(event, self());
return ReceiveBuilder
.match(String.class, msg -> {
persistAsync(String.format("%s-outer-1", msg ), event -> {
sender().tell(event, self());
persistAsync(String.format("%s-inner-1", event), replyToSender);
});
persistAsync(String.format("%s-outer-2", msg ), event -> {
sender().tell(event, self());
persistAsync(String.format("%s-inner-1", event), replyToSender);
});
})
.build();
}
//#nested-persistAsync-persistAsync
void usage(ActorRef persistentActor) {
//#nested-persistAsync-persistAsync-caller
persistentActor.tell("a", self());
persistentActor.tell("b", self());
// order of received messages:
// a
// b
// a-outer-1
// a-outer-2
// b-outer-1
// b-outer-2
// a-inner-1
// a-inner-2
// b-inner-1
// b-inner-2
// which can be seen as the following causal relationship:
// a -> a-outer-1 -> a-outer-2 -> a-inner-1 -> a-inner-2
// b -> b-outer-1 -> b-outer-2 -> b-inner-1 -> b-inner-2
//#nested-persistAsync-persistAsync-caller
}
}
};
static Object o12 = new Object() {
//#view
class MyView extends AbstractPersistentView {
@Override public String persistenceId() { return "some-persistence-id"; }
@Override public String viewId() { return "some-persistence-id-view"; }
public MyView() {
receive(ReceiveBuilder.
match(Object.class, p -> isPersistent(), persistent -> {
// ...
}).build()
);
}
}
//#view
public void usage() {
final ActorSystem system = ActorSystem.create("example");
//#view-update
final ActorRef view = system.actorOf(Props.create(MyView.class));
view.tell(Update.create(true), null);
//#view-update
}
};
}

114
akka-docs/rst/java/code/docs/persistence/LambdaPersistencePluginDocTest.java Normal file
View file

@ -0,0 +1,114 @@
/**
* Copyright (C) 2009-2015 Typesafe Inc. <http://www.typesafe.com>
*/
package docs.persistence;
//#plugin-imports
import akka.dispatch.Futures;
import akka.persistence.*;
import akka.persistence.journal.japi.*;
import akka.persistence.snapshot.japi.*;
//#plugin-imports
import akka.actor.*;
import akka.persistence.journal.leveldb.SharedLeveldbJournal;
import akka.persistence.journal.leveldb.SharedLeveldbStore;
import akka.japi.pf.ReceiveBuilder;
import java.util.ArrayList;
import scala.concurrent.Future;
import java.util.function.Consumer;
import java.util.Optional;
public class LambdaPersistencePluginDocTest {
static Object o1 = new Object() {
final ActorSystem system = null;
//#shared-store-creation
final ActorRef store = system.actorOf(Props.create(SharedLeveldbStore.class), "store");
//#shared-store-creation
//#shared-store-usage
class SharedStorageUsage extends AbstractActor {
@Override
public void preStart() throws Exception {
String path = "akka.tcp://example@127.0.0.1:2552/user/store";
ActorSelection selection = context().actorSelection(path);
selection.tell(new Identify(1), self());
}
public SharedStorageUsage() {
receive(ReceiveBuilder.
match(ActorIdentity.class, ai -> {
if (ai.correlationId().equals(1)) {
ActorRef store = ai.getRef();
if (store != null) {
SharedLeveldbJournal.setStore(store, context().system());
}
}
}).build()
);
}
}
//#shared-store-usage
};
class MySnapshotStore extends SnapshotStore {
@Override
public Future<Optional<SelectedSnapshot>> doLoadAsync(String persistenceId, SnapshotSelectionCriteria criteria) {
return null;
}
@Override
public Future<Void> doSaveAsync(SnapshotMetadata metadata, Object snapshot) {
return null;
}
@Override
public Future<Void> doDeleteAsync(SnapshotMetadata metadata) {
return Futures.successful(null);
}
@Override
public Future<Void> doDeleteAsync(String persistenceId, SnapshotSelectionCriteria criteria) {
return Futures.successful(null);
}
}
class MyAsyncJournal extends AsyncWriteJournal {
//#sync-journal-plugin-api
@Override
public Future<Iterable<Optional<Exception>>> doAsyncWriteMessages(
Iterable<AtomicWrite> messages) {
try {
Iterable<Optional<Exception>> result = new ArrayList<Optional<Exception>>();
// blocking call here...
// result.add(..)
return Futures.successful(result);
} catch (Exception e) {
return Futures.failed(e);
}
}
//#sync-journal-plugin-api
@Override
public Future<Void> doAsyncDeleteMessagesTo(String persistenceId, long toSequenceNr) {
return null;
}
@Override
public Future<Void> doAsyncReplayMessages(String persistenceId, long fromSequenceNr,
long toSequenceNr,
long max,
Consumer<PersistentRepr> replayCallback) {
return null;
}
@Override
public Future<Long> doAsyncReadHighestSequenceNr(String persistenceId, long fromSequenceNr) {
return null;
}
}
}

51
akka-docs/rst/java/code/docs/persistence/PersistenceDocTest.java
View file

@ -32,41 +32,23 @@ public class PersistenceDocTest {
//#recovery-status
}
static Object o1 = new Object() {
class MyActor extends UntypedActor {
ActorRef persistentActor;
public void onReceive(Object message) throws Exception {
}
private void recover() {
//#recover-explicit
persistentActor.tell(Recover.create(), self());
//#recover-explicit
}
}
};
static Object o2 = new Object() {
abstract class MyPersistentActor1 extends UntypedPersistentActor {
//#recover-on-start-disabled
//#recovery-disabled
@Override
public void preStart() {}
//#recover-on-start-disabled
//#recover-on-restart-disabled
@Override
public void preRestart(Throwable reason, Option<Object> message) {}
//#recover-on-restart-disabled
public Recovery recovery() {
return Recovery.none();
}
//#recovery-disabled
}
abstract class MyPersistentActor2 extends UntypedPersistentActor {
//#recover-on-start-custom
//#recovery-custom
@Override
public void preStart() {
self().tell(Recover.create(457L), self());
public Recovery recovery() {
return Recovery.create(457L);
}
//#recover-on-start-custom
//#recovery-custom
}
class MyPersistentActor4 extends UntypedPersistentActor implements PersistentActorMethods {
@ -125,15 +107,6 @@ public class PersistenceDocTest {
}
};
static Object fullyDisabledRecoveryExample = new Object() {
abstract class MyPersistentActor1 extends UntypedPersistentActor {
//#recover-fully-disabled
@Override
public void preStart() { self().tell(Recover.create(0L), self()); }
//#recover-fully-disabled
}
};
static Object atLeastOnceExample = new Object() {
//#at-least-once-example
@ -227,7 +200,7 @@ public class PersistenceDocTest {
if (message instanceof Msg) {
Msg msg = (Msg) message;
// ...
getSender().tell(new Confirm(msg.deliveryId), self());
getSender().tell(new Confirm(msg.deliveryId), getSelf());
} else {
unhandled(message);
}
@ -304,7 +277,7 @@ public class PersistenceDocTest {
private void recover() {
//#snapshot-criteria
persistentActor.tell(Recover.create(SnapshotSelectionCriteria.create(457L,
persistentActor.tell(Recovery.create(SnapshotSelectionCriteria.create(457L,
System.currentTimeMillis())), null);
//#snapshot-criteria
}
@ -541,7 +514,7 @@ public class PersistenceDocTest {
}
};
static Object o12 = new Object() {
static Object o13 = new Object() {
//#view
class MyView extends UntypedPersistentView {
@Override

69
akka-docs/rst/java/lambda-persistence.rst
View file

@ -161,7 +161,7 @@ Identifiers
A persistent actor must have an identifier that doesn't change across different actor incarnations.
The identifier must be defined with the ``persistenceId`` method.
.. includecode:: ../../../akka-samples/akka-sample-persistence-java-lambda/src/main/java/doc/LambdaPersistenceDocTest.java#persistence-id-override
.. includecode:: code/docs/persistence/LambdaPersistenceDocTest.java#persistence-id-override
.. _recovery-java-lambda:
@ -175,46 +175,29 @@ only be received by a persistent actor after recovery completes.
Recovery customization
^^^^^^^^^^^^^^^^^^^^^^
Automated recovery on start can be disabled by overriding ``preStart`` with an empty implementation.
Applications may also customise how recovery is performed by returning a customised ``Recovery`` object
in the ``recovery`` method of a ``AbstractPersistentActor``, for example setting an upper bound to the replay,
which allows the actor to be replayed to a certain point "in the past" instead to its most up to date state:
.. includecode:: ../../../akka-samples/akka-sample-persistence-java-lambda/src/main/java/doc/LambdaPersistenceDocTest.java#recover-on-start-disabled
.. includecode:: code/docs/persistence/LambdaPersistenceDocTest.java#recovery-custom
In this case, a persistent actor must be recovered explicitly by sending it a ``Recover`` message.
Recovery can be disabled by returning ``Recovery.none`` in the ``recovery`` method of a ``PersistentActor``:
.. includecode:: ../../../akka-samples/akka-sample-persistence-java-lambda/src/main/java/doc/LambdaPersistenceDocTest.java#recover-explicit
.. warning::
If ``preStart`` is overridden by an empty implementation, incoming commands will not be processed by the
``PersistentActor`` until it receives a ``Recover`` and finishes recovery.
In order to completely skip recovery, you can signal it with ``Recover.create(0L)``
.. includecode:: ../../../akka-samples/akka-sample-persistence-java-lambda/src/main/java/doc/LambdaPersistenceDocTest.java#recover-fully-disabled
If not overridden, ``preStart`` sends a ``Recover`` message to ``self()``. Applications may also override
``preStart`` to define further ``Recover`` parameters such as an upper sequence number bound, for example.
.. includecode:: ../../../akka-samples/akka-sample-persistence-java-lambda/src/main/java/doc/LambdaPersistenceDocTest.java#recover-on-start-custom
Upper sequence number bounds can be used to recover a persistent actor to past state instead of current state. Automated
recovery on restart can be disabled by overriding ``preRestart`` with an empty implementation.
.. includecode:: ../../../akka-samples/akka-sample-persistence-java-lambda/src/main/java/doc/LambdaPersistenceDocTest.java#recover-on-restart-disabled
.. includecode:: code/docs/persistence/LambdaPersistenceDocTest.java#recovery-disabled
Recovery status
^^^^^^^^^^^^^^^
A persistent actor can query its own recovery status via the methods
.. includecode:: ../../../akka-samples/akka-sample-persistence-java-lambda/src/main/java/doc/LambdaPersistenceDocTest.java#recovery-status
.. includecode:: code/docs/persistence/LambdaPersistenceDocTest.java#recovery-status
Sometimes there is a need for performing additional initialization when the
recovery has completed, before processing any other message sent to the persistent actor.
The persistent actor will receive a special :class:`RecoveryCompleted` message right after recovery
and before any other received messages.
.. includecode:: ../../../akka-samples/akka-sample-persistence-java-lambda/src/main/java/doc/LambdaPersistenceDocTest.java#recovery-completed
.. includecode:: code/docs/persistence/LambdaPersistenceDocTest.java#recovery-completed
If there is a problem with recovering the state of the actor from the journal, ``onReplayFailure``
is called (logging the error by default) and the actor will be stopped.
@ -236,7 +219,7 @@ stash incoming Commands while the Journal is still working on persisting and/or
In the below example, the event callbacks may be called "at any time", even after the next Command has been processed.
The ordering between events is still guaranteed ("evt-b-1" will be sent after "evt-a-2", which will be sent after "evt-a-1" etc.).
.. includecode:: ../../../akka-samples/akka-sample-persistence-java-lambda/src/main/java/doc/LambdaPersistenceDocTest.java#persist-async
.. includecode:: code/docs/persistence/LambdaPersistenceDocTest.java#persist-async
.. note::
In order to implement the pattern known as "*command sourcing*" simply call ``persistAsync`` on all incoming messages right away,
@ -259,12 +242,12 @@ use it for *read* operations, and actions which do not have corresponding events
Using this method is very similar to the persist family of methods, yet it does **not** persist the passed in event.
It will be kept in memory and used when invoking the handler.
.. includecode:: ../../../akka-samples/akka-sample-persistence-java-lambda/src/main/java/doc/LambdaPersistenceDocTest.java#defer
.. includecode:: code/docs/persistence/LambdaPersistenceDocTest.java#defer
Notice that the ``sender()`` is **safe** to access in the handler callback, and will be pointing to the original sender
of the command for which this ``defer`` handler was called.
.. includecode:: ../../../akka-samples/akka-sample-persistence-java-lambda/src/main/java/doc/LambdaPersistenceDocTest.java#defer-caller
.. includecode:: code/docs/persistence/LambdaPersistenceDocTest.java#defer-caller
.. warning::
The callback will not be invoked if the actor is restarted (or stopped) in between the call to
@ -282,11 +265,11 @@ however there are situations where it may be useful. It is important to understa
those situations, as well as their implication on the stashing behaviour (that ``persist()`` enforces). In the following
example two persist calls are issued, and each of them issues another persist inside its callback:
.. includecode:: ../../../akka-samples/akka-sample-persistence-java-lambda/src/main/java/doc/LambdaPersistenceDocTest.java#nested-persist-persist
.. includecode:: code/docs/persistence/LambdaPersistenceDocTest.java#nested-persist-persist
When sending two commands to this ``PersistentActor``, the persist handlers will be executed in the following order:
.. includecode:: ../../../akka-samples/akka-sample-persistence-java-lambda/src/main/java/doc/LambdaPersistenceDocTest.java#nested-persist-persist-caller
.. includecode:: code/docs/persistence/LambdaPersistenceDocTest.java#nested-persist-persist-caller
First the "outer layer" of persist calls is issued and their callbacks applied, after these have successfully completed
the inner callbacks will be invoked (once the events they are persisting have been confirmed to be persisted by the journal).
@ -296,11 +279,11 @@ is extended until all nested ``persist`` callbacks have been handled.
It is also possible to nest ``persistAsync`` calls, using the same pattern:
.. includecode:: ../../../akka-samples/akka-sample-persistence-java-lambda/src/main/java/doc/LambdaPersistenceDocTest.java#nested-persistAsync-persistAsync
.. includecode:: code/docs/persistence/LambdaPersistenceDocTest.java#nested-persistAsync-persistAsync
In this case no stashing is happening, yet the events are still persisted and callbacks executed in the expected order:
.. includecode:: ../../../akka-samples/akka-sample-persistence-java-lambda/src/main/java/doc/LambdaPersistenceDocTest.java#nested-persistAsync-persistAsync-caller
.. includecode:: code/docs/persistence/LambdaPersistenceDocTest.java#nested-persistAsync-persistAsync-caller
While it is possible to nest mixed ``persist`` and ``persistAsync`` with keeping their respective semantics
it is not a recommended practice as it may lead to overly complex nesting.
@ -317,7 +300,7 @@ will most likely fail anyway, since the journal is probably unavailable. It is b
actor and after a back-off timeout start it again. The ``akka.persistence.BackoffSupervisor`` actor
is provided to support such restarts.
.. includecode:: ../../../akka-samples/akka-sample-persistence-java-lambda/src/main/java/doc/LambdaPersistenceDocTest.java#backoff
.. includecode:: code/docs/persistence/LambdaPersistenceDocTest.java#backoff
If persistence of an event is rejected before it is stored, e.g. due to serialization error,
``onPersistRejected`` will be invoked (logging a warning by default) and the actor continues with
@ -372,7 +355,7 @@ Views
Persistent views can be implemented by extending the ``AbstractView`` abstract class, implement the ``persistenceId`` method
and setting the “initial behavior” in the constructor by calling the :meth:`receive` method.
.. includecode:: ../../../akka-samples/akka-sample-persistence-java-lambda/src/main/java/doc/LambdaPersistenceDocTest.java#view
.. includecode:: code/docs/persistence/LambdaPersistenceDocTest.java#view
The ``persistenceId`` identifies the persistent actor from which the view receives journaled messages. It is not necessary
the referenced persistent actor is actually running. Views read messages from a persistent actor's journal directly. When a
@ -394,7 +377,7 @@ The default update interval of all persistent views of an actor system is config
interval for a specific view class or view instance. Applications may also trigger additional updates at
any time by sending a view an ``Update`` message.
.. includecode:: ../../../akka-samples/akka-sample-persistence-java-lambda/src/main/java/doc/LambdaPersistenceDocTest.java#view-update
.. includecode:: code/docs/persistence/LambdaPersistenceDocTest.java#view-update
If the ``await`` parameter is set to ``true``, messages that follow the ``Update`` request are processed when the
incremental message replay, triggered by that update request, completed. If set to ``false`` (default), messages
@ -439,12 +422,12 @@ in context of persistent actors but this is also applicable to persistent views.
Persistent actor can save snapshots of internal state by calling the ``saveSnapshot`` method. If saving of a snapshot
succeeds, the persistent actor receives a ``SaveSnapshotSuccess`` message, otherwise a ``SaveSnapshotFailure`` message
.. includecode:: ../../../akka-samples/akka-sample-persistence-java-lambda/src/main/java/doc/LambdaPersistenceDocTest.java#save-snapshot
.. includecode:: code/docs/persistence/LambdaPersistenceDocTest.java#save-snapshot
During recovery, the persistent actor is offered a previously saved snapshot via a ``SnapshotOffer`` message from
which it can initialize internal state.
.. includecode:: ../../../akka-samples/akka-sample-persistence-java-lambda/src/main/java/doc/LambdaPersistenceDocTest.java#snapshot-offer
.. includecode:: code/docs/persistence/LambdaPersistenceDocTest.java#snapshot-offer
The replayed messages that follow the ``SnapshotOffer`` message, if any, are younger than the offered snapshot.
They finally recover the persistent actor to its current (i.e. latest) state.
@ -452,7 +435,7 @@ They finally recover the persistent actor to its current (i.e. latest) state.
In general, a persistent actor is only offered a snapshot if that persistent actor has previously saved one or more snapshots
and at least one of these snapshots matches the ``SnapshotSelectionCriteria`` that can be specified for recovery.
.. includecode:: ../../../akka-samples/akka-sample-persistence-java-lambda/src/main/java/doc/LambdaPersistenceDocTest.java#snapshot-criteria
.. includecode:: code/docs/persistence/LambdaPersistenceDocTest.java#snapshot-criteria
If not specified, they default to ``SnapshotSelectionCriteria.latest()`` which selects the latest (= youngest) snapshot.
To disable snapshot-based recovery, applications should use ``SnapshotSelectionCriteria.none()``. A recovery where no
@ -509,7 +492,7 @@ between ``deliver`` and ``confirmDelivery`` is possible. The ``deliveryId`` must
of the message, destination actor will send the same``deliveryId`` wrapped in a confirmation message back to the sender.
The sender will then use it to call ``confirmDelivery`` method to complete delivery routine.
.. includecode:: ../../../akka-samples/akka-sample-persistence-java-lambda/src/main/java/doc/LambdaPersistenceDocTest.java#at-least-once-example
.. includecode:: code/docs/persistence/LambdaPersistenceDocTest.java#at-least-once-example
The ``deliveryId`` generated by the persistence module is a strictly monotonically increasing sequence number
without gaps. The same sequence is used for all destinations of the actor, i.e. when sending to multiple
@ -631,7 +614,7 @@ For an example of snapshot store plugin which writes snapshots as individual fil
Applications can provide their own plugins by implementing a plugin API and activate them by configuration.
Plugin development requires the following imports:
.. includecode:: ../../../akka-samples/akka-sample-persistence-java-lambda/src/main/java/doc/LambdaPersistencePluginDocTest.java#plugin-imports
.. includecode:: code/docs/persistence/LambdaPersistencePluginDocTest.java#plugin-imports
Journal plugin API
------------------
@ -644,7 +627,7 @@ A journal plugin extends ``AsyncWriteJournal``.
If the storage backend API only supports synchronous, blocking writes, the methods should be implemented as:
.. includecode:: ../../../akka-samples/akka-sample-persistence-java-lambda/src/main/java/doc/LambdaPersistencePluginDocTest.java#sync-journal-plugin-api
.. includecode:: code/docs/persistence/LambdaPersistencePluginDocTest.java#sync-journal-plugin-api
A journal plugin must also implement the methods defined in ``AsyncRecovery`` for replays and sequence number recovery:
@ -719,7 +702,7 @@ plugin.
This plugin must be initialized by injecting the (remote) ``SharedLeveldbStore`` actor reference. Injection is
done by calling the ``SharedLeveldbJournal.setStore`` method with the actor reference as argument.
.. includecode:: ../../../akka-samples/akka-sample-persistence-java-lambda/src/main/java/doc/LambdaPersistencePluginDocTest.java#shared-store-usage
.. includecode:: code/docs/persistence/PersistencePluginDocTest.java#shared-store-usage
Internal journal commands (sent by persistent actors) are buffered until injection completes. Injection is idempotent
i.e. only the first injection is used.

29
akka-docs/rst/java/persistence.rst
View file

@ -177,32 +177,15 @@ They are cached and received by a persistent actor after recovery phase complete
Recovery customization
^^^^^^^^^^^^^^^^^^^^^^
Automated recovery on start can be disabled by overriding ``preStart`` with an empty or custom implementation.
Applications may also customise how recovery is performed by returning a customised ``Recovery`` object
in the ``recovery`` method of a ``UntypedPersistentActor``, for example setting an upper bound to the replay,
which allows the actor to be replayed to a certain point "in the past" instead to its most up to date state:
.. includecode:: code/docs/persistence/PersistenceDocTest.java#recover-on-start-disabled
.. includecode:: code/docs/persistence/PersistenceDocTest.java#recovery-custom
In this case, a persistent actor must be recovered explicitly by sending it a ``Recover`` message.
Recovery can be disabled by returning ``Recovery.none()`` in the ``recovery`` method of a ``PersistentActor``:
.. includecode:: code/docs/persistence/PersistenceDocTest.java#recover-explicit
.. warning::
If ``preStart`` is overridden by an empty implementation, incoming commands will not be processed by the
``PersistentActor`` until it receives a ``Recover`` and finishes recovery.
In order to completely skip recovery, you can signal it with ``Recover.create(0L)``
.. includecode:: code/docs/persistence/PersistenceDocTest.java#recover-fully-disabled
If not overridden, ``preStart`` sends a ``Recover`` message to ``getSelf()``. Applications may also override
``preStart`` to define further ``Recover`` parameters such as an upper sequence number bound, for example.
.. includecode:: code/docs/persistence/PersistenceDocTest.java#recover-on-start-custom
Upper sequence number bounds can be used to recover a persistent actor to past state instead of current state. Automated
recovery on restart can be disabled by overriding ``preRestart`` with an empty implementation.
.. includecode:: code/docs/persistence/PersistenceDocTest.java#recover-on-restart-disabled
.. includecode:: code/docs/persistence/PersistenceDocTest.java#recovery-disabled
Recovery status
^^^^^^^^^^^^^^^