AtLeastOnceTransmission.scala

import akka.actor.ActorPath

trait AtLeastOnceTransmission[A] extends akka.persistence.AtLeastOnceDelivery {
  private var unconfirmed = Map[A, Set[Long]]()

  def transmit(destination: ActorPath, transmissionId: A, message: Any): Unit = {
    deliver(destination, { deliveryId ⇒
      unconfirmed = add(unconfirmed, transmissionId, deliveryId)
      message
    })
  }

  def confirmTransmission(transmissionId: A): Boolean = {
    unconfirmed.get(transmissionId) match {
      case Some(deliveryIds) ⇒
        unconfirmed = unconfirmed - transmissionId
        deliveryIds.forall(super.confirmDelivery)
      case None ⇒ false
    }
  }

  def add[B, C](map: Map[B, Set[C]], key: B, value: C): Map[B, Set[C]] = map.updated(key, map.getOrElse(key, Set[C]()) + value)
}