Akka/Scala example source code file (AsyncWriteProxy.scala)

This example Akka source code file (AsyncWriteProxy.scala) is included in my "Source Code Warehouse" project. The intent of this project is to help you more easily find Akka and Scala source code examples by using tags.

All credit for the original source code belongs to akka.io; I'm just trying to make examples easier to find. (For my Scala work, see my Scala examples and tutorials.)

Akka tags/keywords

akka, asyncreplaytimeoutexception, asyncwriteproxy, boolean, collection, concurrent, duration, future, long, persistence, persistentrepr, promise, replaysuccess, serialversionuid, string, time

The AsyncWriteProxy.scala Akka example source code

/**
 * Copyright (C) 2009-2014 Typesafe Inc. <http://www.typesafe.com>
 */

package akka.persistence.journal

import scala.collection.immutable
import scala.concurrent._
import scala.concurrent.duration.Duration
import scala.language.postfixOps

import akka.AkkaException
import akka.actor._
import akka.pattern.ask
import akka.persistence._
import akka.util._

/**
 * INTERNAL API.
 *
 * A journal that delegates actual storage to a target actor. For testing only.
 */
private[persistence] trait AsyncWriteProxy extends AsyncWriteJournal with Stash {
  import AsyncWriteProxy._
  import AsyncWriteTarget._

  private val initialized = super.receive
  private var store: ActorRef = _

  override def receive = {
    case SetStore(ref) ⇒
      store = ref
      unstashAll()
      context.become(initialized)
    case _ ⇒ stash()
  }

  implicit def timeout: Timeout

  def asyncWriteMessages(messages: immutable.Seq[PersistentRepr]): Future[Unit] =
    (store ? WriteMessages(messages)).mapTo[Unit]

  def asyncWriteConfirmations(confirmations: immutable.Seq[PersistentConfirmation]): Future[Unit] =
    (store ? WriteConfirmations(confirmations)).mapTo[Unit]

  def asyncDeleteMessages(messageIds: immutable.Seq[PersistentId], permanent: Boolean): Future[Unit] =
    (store ? DeleteMessages(messageIds, permanent)).mapTo[Unit]

  def asyncDeleteMessagesTo(persistenceId: String, toSequenceNr: Long, permanent: Boolean): Future[Unit] =
    (store ? DeleteMessagesTo(persistenceId, toSequenceNr, permanent)).mapTo[Unit]

  def asyncReplayMessages(persistenceId: String, fromSequenceNr: Long, toSequenceNr: Long, max: Long)(replayCallback: (PersistentRepr) ⇒ Unit): Future[Unit] = {
    val replayCompletionPromise = Promise[Unit]
    val mediator = context.actorOf(Props(classOf[ReplayMediator], replayCallback, replayCompletionPromise, timeout.duration).withDeploy(Deploy.local))
    store.tell(ReplayMessages(persistenceId, fromSequenceNr, toSequenceNr, max), mediator)
    replayCompletionPromise.future
  }

  def asyncReadHighestSequenceNr(persistenceId: String, fromSequenceNr: Long): Future[Long] =
    (store ? ReadHighestSequenceNr(persistenceId, fromSequenceNr)).mapTo[Long]
}

/**
 * INTERNAL API.
 */
private[persistence] object AsyncWriteProxy {
  final case class SetStore(ref: ActorRef)
}

/**
 * INTERNAL API.
 */
private[persistence] object AsyncWriteTarget {
  @SerialVersionUID(1L)
  final case class WriteMessages(messages: immutable.Seq[PersistentRepr])

  @SerialVersionUID(1L)
  final case class WriteConfirmations(confirmations: immutable.Seq[PersistentConfirmation])

  @SerialVersionUID(1L)
  final case class DeleteMessages(messageIds: immutable.Seq[PersistentId], permanent: Boolean)

  @SerialVersionUID(1L)
  final case class DeleteMessagesTo(persistenceId: String, toSequenceNr: Long, permanent: Boolean)

  @SerialVersionUID(1L)
  final case class ReplayMessages(persistenceId: String, fromSequenceNr: Long, toSequenceNr: Long, max: Long)

  @SerialVersionUID(1L)
  case object ReplaySuccess

  @SerialVersionUID(1L)
  final case class ReplayFailure(cause: Throwable)

  @SerialVersionUID(1L)
  final case class ReadHighestSequenceNr(persistenceId: String, fromSequenceNr: Long)
}

/**
 * Thrown if replay inactivity exceeds a specified timeout.
 */
@SerialVersionUID(1L)
class AsyncReplayTimeoutException(msg: String) extends AkkaException(msg)

private class ReplayMediator(replayCallback: PersistentRepr ⇒ Unit, replayCompletionPromise: Promise[Unit], replayTimeout: Duration) extends Actor {
  import AsyncWriteTarget._

  context.setReceiveTimeout(replayTimeout)

  def receive = {
    case p: PersistentRepr ⇒ replayCallback(p)
    case ReplaySuccess ⇒
      replayCompletionPromise.success(())
      context.stop(self)
    case ReplayFailure(cause) ⇒
      replayCompletionPromise.failure(cause)
      context.stop(self)
    case ReceiveTimeout ⇒
      replayCompletionPromise.failure(new AsyncReplayTimeoutException(s"replay timed out after ${replayTimeout.toSeconds} seconds inactivity"))
      context.stop(self)
  }
}