Persistence

Changes in Akka 2.3.4

In Akka 2.3.4 several of the concepts of the earlier versions were collapsed and simplified. In essence; Processor and EventsourcedProcessor are replaced by PersistentActor. Channel and PersistentChannel are replaced by AtLeastOnceDelivery. View is replaced by PersistentView.

See full details of the changes in the Migration Guide Akka Persistence (experimental) 2.3.3 to 2.3.4 (and 2.4.x). The old classes are still included, and deprecated, for a while to make the transition smooth. In case you need the old documentation it is located here.

Dependencies

Akka persistence is a separate jar file. Make sure that you have the following dependency in your project:

<dependency>
  <groupId>com.typesafe.akka</groupId>
  <artifactId>akka-persistence-experimental_2.10</artifactId>
  <version>2.3.16</version>
</dependency>

Architecture

• UntypedPersistentActor: Is a persistent, stateful actor. It is able to persist events to a journal and can react to them in a thread-safe manner. It can be used to implement both command as well as event sourced actors. When a persistent actor is started or restarted, journaled messages are replayed to that actor, so that it can recover internal state from these messages.
• UntypedPersistentView: A view is a persistent, stateful actor that receives journaled messages that have been written by another persistent actor. A view itself does not journal new messages, instead, it updates internal state only from a persistent actor's replicated message stream.
• UntypedPersistentActorAtLeastOnceDelivery: To send messages with at-least-once delivery semantics to destinations, also in case of sender and receiver JVM crashes.
• Journal: A journal stores the sequence of messages sent to a persistent actor. An application can control which messages are journaled and which are received by the persistent actor without being journaled. The storage backend of a journal is pluggable. The default journal storage plugin writes to the local filesystem, replicated journals are available as Community plugins.
• Snapshot store: A snapshot store persists snapshots of a persistent actor's or a view's internal state. Snapshots are used for optimizing recovery times. The storage backend of a snapshot store is pluggable. The default snapshot storage plugin writes to the local filesystem.

Event sourcing

The basic idea behind Event Sourcing is quite simple. A persistent actor receives a (non-persistent) command which is first validated if it can be applied to the current state. Here, validation can mean anything, from simple inspection of a command message's fields up to a conversation with several external services, for example. If validation succeeds, events are generated from the command, representing the effect of the command. These events are then persisted and, after successful persistence, used to change the actor's state. When the persistent actor needs to be recovered, only the persisted events are replayed of which we know that they can be successfully applied. In other words, events cannot fail when being replayed to a persistent actor, in contrast to commands. Event sourced actors may of course also process commands that do not change application state, such as query commands, for example.

Akka persistence supports event sourcing with the UntypedPersistentActor abstract class. An actor that extends this class uses the persist method to persist and handle events. The behavior of an UntypedPersistentActor is defined by implementing receiveRecover and receiveCommand. This is demonstrated in the following example.

import akka.actor.ActorRef;
import akka.actor.ActorSystem;
import akka.actor.Props;
import akka.japi.Procedure;
import akka.persistence.SnapshotOffer;
import akka.persistence.UntypedPersistentActor;

import java.io.Serializable;
import java.util.ArrayList;
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 ExamplePersistentActor extends UntypedPersistentActor {
    @Override
    public String persistenceId() { return "sample-id-1"; }

    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();
            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);
                    }
                }
            });
        } 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);
        } else {
          unhandled(msg);
        }
    }
}

The example defines two data types, Cmd and Evt to represent commands and events, respectively. The state of the ExamplePersistentActor is a list of persisted event data contained in ExampleState.

The persistent actor's onReceiveRecover method defines how state is updated during recovery by handling Evt and SnapshotOffer messages. The persistent actor's onReceiveCommand method is a command handler. In this example, a command is handled by generating two events which are then persisted and handled. Events are persisted by calling persist with an event (or a sequence of events) as first argument and an event handler as second argument.

The persist method persists events asynchronously and the event handler is executed for successfully persisted events. Successfully persisted events are internally sent back to the persistent actor as individual messages that trigger event handler executions. An event handler may close over persistent actor state and mutate it. The sender of a persisted event is the sender of the corresponding command. This allows event handlers to reply to the sender of a command (not shown).

The main responsibility of an event handler is changing persistent actor state using event data and notifying others about successful state changes by publishing events.

When persisting events with persist it is guaranteed that the persistent actor will not receive further commands between the persist call and the execution(s) of the associated event handler. This also holds for multiple persist calls in context of a single command.

The easiest way to run this example yourself is to download Typesafe Activator and open the tutorial named Akka Persistence Samples with Java. It contains instructions on how to run the PersistentActorExample.

@Override
public String persistenceId() {
    return "my-stable-persistence-id";
}

@Override
public void preStart() {}

processor.tell(Recover.create(), null);

@Override
public void preStart() {
    getSelf().tell(Recover.create(457L), null);
}

@Override
public void preRestart(Throwable reason, Option<Object> message) {}

public boolean recoveryRunning();
public boolean recoveryFinished();

          
            @Override
            public void onReceiveRecover(Object message) {
	      if (message instanceof RecoveryCompleted) {
	          recoveryCompleted();
	      }
              // ...
            }
            
            @Override
            public void onReceiveCommand(Object message) throws Exception {
                if (message instanceof String) {
                  // ...
                } else {
                    unhandled(message);
                }
            }
            
            private void recoveryCompleted() {
                // perform init after recovery, before any other messages
                // ...
            }
            

class MyPersistentActor extends UntypedPersistentActor {
    @Override
    public String persistenceId() { return "some-persistence-id"; }

    @Override
    public void onReceiveRecover(Object msg) {
        // handle recovery here
    }

    @Override
    public void onReceiveCommand(Object msg) {
        sender().tell(msg, getSelf());

        persistAsync(String.format("evt-%s-1", msg), new Procedure<String>(){
            @Override
            public void apply(String event) throws Exception {
                sender().tell(event, self());
            }
        });
        persistAsync(String.format("evt-%s-2", msg), new Procedure<String>(){
            @Override
            public void apply(String event) throws Exception {
                sender().tell(event, self());
            }
        });
    }
}

class MyPersistentActor extends UntypedPersistentActor {
    @Override
    public String persistenceId() { return "some-persistence-id"; }

    @Override
    public void onReceiveRecover(Object msg) {
        // handle recovery here
    }

    @Override
    public void onReceiveCommand(Object msg) {
        final Procedure<String> replyToSender = new Procedure<String>() {
            @Override
            public void apply(String event) throws Exception {
                sender().tell(event, self());
            }
        };

        persistAsync(String.format("evt-%s-1", msg), replyToSender);
        persistAsync(String.format("evt-%s-2", msg), replyToSender);
        defer(String.format("evt-%s-3", msg), replyToSender);
    }
}

final ActorRef processor = system.actorOf(Props.create(MyPersistentActor.class));
processor.tell("a", null);
processor.tell("b", null);

// order of received messages:
// a
// b
// evt-a-1
// evt-a-2
// evt-a-3
// evt-b-1
// evt-b-2
// evt-b-3

akka.persistence.journal.max-message-batch-size = 200

Persistent Views

Persistent views can be implemented by extending the UntypedPersistentView trait and implementing the onReceive and the persistenceId methods.

class MyView extends UntypedPersistentView {
    @Override
    public String persistenceId() { return "some-persistence-id"; }
    
    @Override
    public String viewId() { return "my-stable-persistence-view-id"; }

    @Override
    public void onReceive(Object message) throws Exception {
        if (isPersistent()) {
            // handle message from Journal...
        } else if (message instanceof String) {
            // handle message from user...
        } else {
          unhandled(message);
        }
    }
}

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 persistent actor is started later and begins to write new messages, the corresponding view is updated automatically, by default.

It is possible to determine if a message was sent from the Journal or from another actor in user-land by calling the isPersistent method. Having that said, very often you don't need this information at all and can simply apply the same logic to both cases (skip the if isPersistent check).

akka.persistence.view.auto-update-interval = 5s

final ActorRef view = system.actorOf(Props.create(MyView.class));
view.tell(Update.create(true), null);

akka.persistence.view.auto-update = off

Snapshots

Snapshots can dramatically reduce recovery times of persistent actor and views. The following discusses snapshots in context of persistent actor 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

class MyProcessor extends UntypedProcessor {
    private Object state;

    @Override
    public void onReceive(Object message) throws Exception {
        if (message.equals("snap")) {
            saveSnapshot(state);
        } else if (message instanceof SaveSnapshotSuccess) {
            SnapshotMetadata metadata = ((SaveSnapshotSuccess)message).metadata();
            // ...
        } else if (message instanceof SaveSnapshotFailure) {
            SnapshotMetadata metadata = ((SaveSnapshotFailure)message).metadata();
            // ...
        }
    }
}

During recovery, the persistent actor is offered a previously saved snapshot via a SnapshotOffer message from which it can initialize internal state.

class MyProcessor extends UntypedProcessor {
    private Object state;

    @Override
    public void onReceive(Object message) throws Exception {
        if (message instanceof SnapshotOffer) {
            state = ((SnapshotOffer)message).snapshot();
            // ...
        } else if (message instanceof Persistent) {
            // ...
        }
    }
}

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.

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.

processor.tell(Recover.create(SnapshotSelectionCriteria.create(457L, System.currentTimeMillis())), null);

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 saved snapshot matches the specified SnapshotSelectionCriteria will replay all journaled messages.

At-Least-Once Delivery

To send messages with at-least-once delivery semantics to destinations you can extend the UntypedPersistentActorWithAtLeastOnceDelivery class instead of UntypedPersistentActor on the sending side. It takes care of re-sending messages when they have not been confirmed within a configurable timeout.

Use the deliver method to send a message to a destination. Call the confirmDelivery method when the destination has replied with a confirmation message.

class Msg implements Serializable {
  public final long deliveryId;
  public final String s;
  
  public Msg(long deliveryId, String s) {
    this.deliveryId = deliveryId;
    this.s = s;
  }
}

class Confirm implements Serializable {
  public final long deliveryId;
  
  public Confirm(long deliveryId) {
    this.deliveryId = deliveryId;
  }
}

  
class MsgSent implements Serializable {
  public final String s;
  
  public MsgSent(String s) {
    this.s = s;
  }
}
class MsgConfirmed implements Serializable {
  public final long deliveryId;
  
  public MsgConfirmed(long deliveryId) {
    this.deliveryId = deliveryId;
  }
}

class MyPersistentActor extends UntypedPersistentActorWithAtLeastOnceDelivery {
  private final ActorPath destination;

  public MyPersistentActor(ActorPath destination) {
      this.destination = destination;
  }

  public void onReceiveCommand(Object message) {
    if (message instanceof String) {
      String s = (String) message;
      persist(new MsgSent(s), new Procedure<MsgSent>() {
        public void apply(MsgSent evt) {
          updateState(evt);
        }
      });
    } else if (message instanceof Confirm) {
      Confirm confirm = (Confirm) message;
      persist(new MsgConfirmed(confirm.deliveryId), new Procedure<MsgConfirmed>() {
        public void apply(MsgConfirmed evt) {
          updateState(evt);
        }
      });
    } else {
      unhandled(message);
    }
  }
  
  public void onReceiveRecover(Object event) {
    updateState(event);
  }
  
  void updateState(Object event) {
    if (event instanceof MsgSent) {
      final MsgSent evt = (MsgSent) event;
      deliver(destination, new Function<Long, Object>() {
        public Object apply(Long deliveryId) {
          return new Msg(deliveryId, evt.s);
        }
      });
    } else if (event instanceof MsgConfirmed) {
      final MsgConfirmed evt = (MsgConfirmed) event;
      confirmDelivery(evt.deliveryId);
    }
  }
}

class MyDestination extends UntypedActor {
  public void onReceive(Object message) throws Exception {
    if (message instanceof Msg) {
      Msg msg = (Msg) message;
      // ...
      getSender().tell(new Confirm(msg.deliveryId), getSelf());
    } else {
      unhandled(message);
    }
  }
}

Correlation between deliver and confirmDelivery is performed with the deliveryId that is provided as parameter to the deliveryIdToMessage function. The deliveryId is typically passed in the message to the destination, which replies with a message containing the same deliveryId.

The deliveryId 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 destinations the destinations will see gaps in the sequence if no translation is performed.

The UntypedPersistentActorWithAtLeastOnceDelivery class has a state consisting of unconfirmed messages and a sequence number. It does not store this state itself. You must persist events corresponding to the deliver and confirmDelivery invocations from your PersistentActor so that the state can be restored by calling the same methods during the recovery phase of the PersistentActor. Sometimes these events can be derived from other business level events, and sometimes you must create separate events. During recovery calls to deliver will not send out the message, but it will be sent later if no matching confirmDelivery was performed.

Support for snapshots is provided by getDeliverySnapshot and setDeliverySnapshot. The AtLeastOnceDeliverySnapshot contains the full delivery state, including unconfirmed messages. If you need a custom snapshot for other parts of the actor state you must also include the AtLeastOnceDeliverySnapshot. It is serialized using protobuf with the ordinary Akka serialization mechanism. It is easiest to include the bytes of the AtLeastOnceDeliverySnapshot as a blob in your custom snapshot.

The interval between redelivery attempts is defined by the redeliverInterval method. The default value can be configured with the akka.persistence.at-least-once-delivery.redeliver-interval configuration key. The method can be overridden by implementation classes to return non-default values.

After a number of delivery attempts a AtLeastOnceDelivery.UnconfirmedWarning message will be sent to self. The re-sending will still continue, but you can choose to call confirmDelivery to cancel the re-sending. The number of delivery attempts before emitting the warning is defined by the warnAfterNumberOfUnconfirmedAttempts method. The default value can be configured with the akka.persistence.at-least-once-delivery.warn-after-number-of-unconfirmed-attempts configuration key. The method can be overridden by implementation classes to return non-default values.

The UntypedPersistentActorWithAtLeastOnceDelivery class holds messages in memory until their successful delivery has been confirmed. The limit of maximum number of unconfirmed messages that the actor is allowed to hold in memory is defined by the maxUnconfirmedMessages method. If this limit is exceed the deliver method will not accept more messages and it will throw AtLeastOnceDelivery.MaxUnconfirmedMessagesExceededException. The default value can be configured with the akka.persistence.at-least-once-delivery.max-unconfirmed-messages configuration key. The method can be overridden by implementation classes to return non-default values.

Storage plugins

Storage backends for journals and snapshot stores are pluggable in Akka persistence. The default journal plugin writes messages to LevelDB (see Local LevelDB journal). The default snapshot store plugin writes snapshots as individual files to the local filesystem (see Local snapshot store). Applications can provide their own plugins by implementing a plugin API and activate them by configuration. Plugin development requires the following imports:

import akka.actor.*;
import akka.japi.Option;
import akka.japi.Procedure;
import akka.persistence.*;
import akka.persistence.japi.journal.JavaJournalSpec;
import akka.persistence.japi.snapshot.JavaSnapshotStoreSpec;
import akka.persistence.journal.japi.AsyncWriteJournal;
import akka.persistence.journal.leveldb.SharedLeveldbJournal;
import akka.persistence.journal.leveldb.SharedLeveldbStore;
import akka.persistence.snapshot.japi.SnapshotStore;
import com.typesafe.config.Config;
import com.typesafe.config.ConfigFactory;
import org.iq80.leveldb.util.FileUtils;
import scala.concurrent.Future;

import java.io.File;
import java.util.ArrayList;
import java.util.List;

/**
 * Java API, Plugin API: synchronously writes a batch of persistent messages to the
 * journal. The batch write must be atomic i.e. either all persistent messages in the
 * batch are written or none.
 */
void doWriteMessages(Iterable<PersistentRepr> messages);

/**
 * Java API, Plugin API: synchronously writes a batch of delivery confirmations to
 * the journal.
 * 
 * @deprecated doWriteConfirmations will be removed, since Channels will be removed (since 2.3.4)
 */
@Deprecated void doWriteConfirmations(Iterable<PersistentConfirmation> confirmations);

/**
 * Java API, Plugin API: synchronously deletes messages identified by `messageIds`
 * from the journal. If `permanent` is set to `false`, the persistent messages are
 * marked as deleted, otherwise they are permanently deleted.
 * 
 * @deprecated doDeleteMessages will be removed (since 2.3.4)
 */
@Deprecated void doDeleteMessages(Iterable<PersistentId> messageIds, boolean permanent);

/**
 * Java API, Plugin API: synchronously deletes all persistent messages up to
 * `toSequenceNr`. If `permanent` is set to `false`, the persistent messages are
 * marked as deleted, otherwise they are permanently deleted.
 *
 * @see AsyncRecoveryPlugin
 */
void doDeleteMessagesTo(String persistenceId, long toSequenceNr, boolean permanent);

/**
 * Java API, Plugin API: synchronously writes a batch of persistent messages to the
 * journal. The batch write must be atomic i.e. either all persistent messages in the
 * batch are written or none.
 */
Future<Void> doAsyncWriteMessages(Iterable<PersistentRepr> messages);

/**
 * Java API, Plugin API: synchronously writes a batch of delivery confirmations to
 * the journal.
 * 
 * @deprecated doAsyncWriteConfirmations will be removed, since Channels will be removed (since 2.3.4)
 */
@Deprecated Future<Void> doAsyncWriteConfirmations(Iterable<PersistentConfirmation> confirmations);

/**
 * Java API, Plugin API: synchronously deletes messages identified by `messageIds`
 * from the journal. If `permanent` is set to `false`, the persistent messages are
 * marked as deleted, otherwise they are permanently deleted.
 * 
 * @deprecated doAsyncDeleteMessages will be removed (since 2.3.4)
 */
@Deprecated Future<Void> doAsyncDeleteMessages(Iterable<PersistentId> messageIds, boolean permanent);

/**
 * Java API, Plugin API: synchronously deletes all persistent messages up to
 * `toSequenceNr`. If `permanent` is set to `false`, the persistent messages are
 * marked as deleted, otherwise they are permanently deleted.
 *
 * @see AsyncRecoveryPlugin
 */
Future<Void> doAsyncDeleteMessagesTo(String persistenceId, long toSequenceNr, boolean permanent);

/**
 * Java API, Plugin API: asynchronously replays persistent messages.
 * Implementations replay a message by calling `replayCallback`. The returned
 * future must be completed when all messages (matching the sequence number
 * bounds) have been replayed. The future must be completed with a failure if
 * any of the persistent messages could not be replayed.
 *
 * The `replayCallback` must also be called with messages that have been marked
 * as deleted. In this case a replayed message's `deleted` method must return
 * `true`.
 *
 * The channel ids of delivery confirmations that are available for a replayed
 * message must be contained in that message's `confirms` sequence.
 *
 * @param persistenceId processor id.
 * @param fromSequenceNr sequence number where replay should start (inclusive).
 * @param toSequenceNr sequence number where replay should end (inclusive).
 * @param max maximum number of messages to be replayed.
 * @param replayCallback called to replay a single message. Can be called from any
 *                       thread.
 */
Future<Void> doAsyncReplayMessages(String persistenceId, long fromSequenceNr, long toSequenceNr, long max, Procedure<PersistentRepr> replayCallback);

/**
 * Java API, Plugin API: asynchronously reads the highest stored sequence number
 * for the given `persistenceId`.
 *
 * @param persistenceId processor id.
 * @param fromSequenceNr hint where to start searching for the highest sequence
 *                       number.
 */
Future<Long> doAsyncReadHighestSequenceNr(String persistenceId, long fromSequenceNr);

# Path to the journal plugin to be used
akka.persistence.journal.plugin = "my-journal"

# My custom journal plugin
my-journal {
  # Class name of the plugin.
  class = "docs.persistence.MyJournal"
  # Dispatcher for the plugin actor.
  plugin-dispatcher = "akka.actor.default-dispatcher"
}

/**
 * Java API, Plugin API: asynchronously loads a snapshot.
 *
 * @param persistenceId processor id.
 * @param criteria selection criteria for loading.
 */
Future<Option<SelectedSnapshot>> doLoadAsync(String persistenceId, SnapshotSelectionCriteria criteria);

/**
 * Java API, Plugin API: asynchronously saves a snapshot.
 *
 * @param metadata snapshot metadata.
 * @param snapshot snapshot.
 */
Future<Void> doSaveAsync(SnapshotMetadata metadata, Object snapshot);

/**
 * Java API, Plugin API: called after successful saving of a snapshot.
 *
 * @param metadata snapshot metadata.
 */
void onSaved(SnapshotMetadata metadata) throws Exception;

/**
 * Java API, Plugin API: deletes the snapshot identified by `metadata`.
 *
 * @param metadata snapshot metadata.
 */
void doDelete(SnapshotMetadata metadata) throws Exception;

/**
 * Java API, Plugin API: deletes all snapshots matching `criteria`.
 *
 * @param persistenceId processor id.
 * @param criteria selection criteria for deleting.
 */
void doDelete(String persistenceId, SnapshotSelectionCriteria criteria) throws Exception;

# Path to the snapshot store plugin to be used
akka.persistence.snapshot-store.plugin = "my-snapshot-store"

# My custom snapshot store plugin
my-snapshot-store {
  # Class name of the plugin.
  class = "docs.persistence.MySnapshotStore"
  # Dispatcher for the plugin actor.
  plugin-dispatcher = "akka.persistence.dispatchers.default-plugin-dispatcher"
}

<dependency>
  <groupId>com.typesafe.akka</groupId>
  <artifactId>akka-persistence-tck-experimental_${scala.version}</artifactId>
  <version>2.3.5</version>
  <scope>test</scope>
</dependency>

class MyJournalSpecTest extends JavaJournalSpec {

    public MyJournalSpecTest() {
        super(ConfigFactory.parseString(
                "persistence.journal.plugin = " +
                "\"akka.persistence.journal.leveldb-shared\""));
    }
}

class MySnapshotStoreTest extends JavaSnapshotStoreSpec {

    public MySnapshotStoreTest() {
        super(ConfigFactory.parseString(
                "akka.persistence.snapshot-store.plugin = " +
                "\"akka.persistence.snapshot-store.local\""));
    }
}

class MyJournalSpecTest extends JavaJournalSpec {

    List<File> storageLocations = new ArrayList<File>();

    public MyJournalSpecTest() {
        super(ConfigFactory.parseString(
                "persistence.journal.plugin = " +
                "\"akka.persistence.journal.leveldb-shared\""));

        Config config = system().settings().config();
        storageLocations.add(new File(
            config.getString("akka.persistence.journal.leveldb.dir")));
        storageLocations.add(new File(
            config.getString("akka.persistence.snapshot-store.local.dir")));
    }

    @Override
    public void beforeAll() {
        for (File storageLocation : storageLocations) {
            FileUtils.deleteRecursively(storageLocation);
        }
        super.beforeAll();
    }

    @Override
    public void afterAll() {
        super.afterAll();
        for (File storageLocation : storageLocations) {
            FileUtils.deleteRecursively(storageLocation);
        }
    }
}

Pre-packaged plugins

akka.persistence.journal.leveldb.dir = "target/journal"

final ActorRef store = system.actorOf(Props.create(SharedLeveldbStore.class), "store");

akka.persistence.journal.leveldb-shared.store.dir = "target/shared"

akka.persistence.journal.plugin = "akka.persistence.journal.leveldb-shared"

class SharedStorageUsage extends UntypedActor {
    @Override
    public void preStart() throws Exception {
        String path = "akka.tcp://[email protected]:2552/user/store";
        ActorSelection selection = getContext().actorSelection(path);
        selection.tell(new Identify(1), getSelf());
    }

    @Override
    public void onReceive(Object message) throws Exception {
        if (message instanceof ActorIdentity) {
            ActorIdentity identity = (ActorIdentity) message;
            if (identity.correlationId().equals(1)) {
                ActorRef store = identity.getRef();
                if (store != null) {
                    SharedLeveldbJournal.setStore(store, getContext().system());
                }
            }
        }
    }
}

akka.persistence.snapshot-store.local.dir = "target/snapshots"

Custom serialization

Serialization of snapshots and payloads of Persistent messages is configurable with Akka's Serialization infrastructure. For example, if an application wants to serialize

• payloads of type MyPayload with a custom MyPayloadSerializer and
• snapshots of type MySnapshot with a custom MySnapshotSerializer

it must add

akka.actor {
  serializers {
    my-payload = "docs.persistence.MyPayloadSerializer"
    my-snapshot = "docs.persistence.MySnapshotSerializer"
  }
  serialization-bindings {
    "docs.persistence.MyPayload" = my-payload
    "docs.persistence.MySnapshot" = my-snapshot
  }
}

to the application configuration. If not specified, a default serializer is used.

Testing

When running tests with LevelDB default settings in sbt, make sure to set fork := true in your sbt project otherwise, you'll see an UnsatisfiedLinkError. Alternatively, you can switch to a LevelDB Java port by setting

akka.persistence.journal.leveldb.native = off

or

akka.persistence.journal.leveldb-shared.store.native = off

in your Akka configuration. The LevelDB Java port is for testing purposes only.

Configuration

There are several configuration properties for the persistence module, please refer to the reference configuration.