Durable Mailboxes

my-dispatcher {
  mailbox-type = akka.actor.mailbox.filebased.FileBasedMailboxType
}

import akka.actor.Props;
import akka.actor.ActorRef;
 
    ActorRef myActor = system.actorOf(Props.create(MyUntypedActor.class).
        withDispatcher("my-dispatcher"), "myactor");

import scala.Option;
import com.typesafe.config.Config;
import akka.actor.ActorRef;
import akka.actor.ActorSystem;
import akka.actor.ExtendedActorSystem;
import akka.dispatch.MailboxType;
import akka.dispatch.MessageQueue;
 
public class MyDurableMailboxType implements MailboxType {
 
  public MyDurableMailboxType(ActorSystem.Settings settings, Config config) {
  }
 
  @Override 
  public MessageQueue create(Option<ActorRef> owner,
      Option<ActorSystem> system) {
    if (owner.isEmpty())
      throw new IllegalArgumentException("requires an owner " +
          "(i.e. does not work with BalancingDispatcher)");
    return new MyDurableMessageQueue(owner.get(), 
      (ExtendedActorSystem) system.get());
  }
}

import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.Callable;
import scala.concurrent.duration.Duration;
import akka.actor.ActorRef;
import akka.actor.ExtendedActorSystem;
import akka.actor.mailbox.DurableMessageQueueWithSerialization;
import akka.dispatch.Envelope;
import akka.dispatch.MessageQueue;
import akka.pattern.CircuitBreaker;
 
public class MyDurableMessageQueue extends DurableMessageQueueWithSerialization {
 
  public MyDurableMessageQueue(ActorRef owner, ExtendedActorSystem system) {
    super(owner, system);
  }
 
  private final QueueStorage storage = new QueueStorage();
  // A real-world implementation would use configuration to set the last 
  // three parameters below
  private final CircuitBreaker breaker = CircuitBreaker.create(system().scheduler(),
    5, Duration.create(30, "seconds"), Duration.create(1, "minute"));
 
  @Override
  public void enqueue(ActorRef receiver, final Envelope envelope) {
    breaker.callWithSyncCircuitBreaker(new Callable<Object>() {
      @Override
      public Object call() {
        byte[] data = serialize(envelope);
        storage.push(data);
        return null;
      }
    });
  }
 
  @Override
  public Envelope dequeue() {
    return breaker.callWithSyncCircuitBreaker(new Callable<Envelope>() {
      @Override
      public Envelope call() {
        byte[] data = storage.pull();
        if (data == null)
          return null;
        else
          return deserialize(data);
      }
    });
  }
 
  @Override
  public boolean hasMessages() {
    return breaker.callWithSyncCircuitBreaker(new Callable<Boolean>() {
      @Override
      public Boolean call() {
        return !storage.isEmpty();
      }
    });
  }
 
  @Override
  public int numberOfMessages() {
    return breaker.callWithSyncCircuitBreaker(new Callable<Integer>() {
      @Override
      public Integer call() {
        return storage.size();
      }
    });
  }
 
  /**
   * Called when the mailbox is disposed.
   * An ordinary mailbox would send remaining messages to deadLetters,
   * but the purpose of a durable mailbox is to continue
   * with the same message queue when the actor is started again.
   */
  @Override
  public void cleanUp(ActorRef owner, MessageQueue deadLetters) {}
 
  // dummy queue storage ...
}