Akka producer actors best practices - Part 2

CorrelationIdentifier.scala

httpProducer ! Message("test", Map(Message.MessageExchangeId -> 123))

JmsForwardingProducer.scala

class JmsForwardingProducer(target: ActorRef) extends Actor with Producer {
  def endpointUri = "jms:queue:test"
  override def oneway = true
  override def receiveAfterProduce = {
    case msg => target forward msg
  }
}

class ForwardReceiver extends Actor {
  def receive = {
    case message: Message => self.reply("done")
    case failure: Failure => self.reply("message could not be added to queue: %s" format failure.cause.getMessage)
  }
}

val receiver = Actor.actorOf[ForwardReceiver].start
val producer = Actor.actorOf(new JmsForwardingProducer(receiver)).start

producer !! "test" // gets a response from ForwardReceiver

SomeProducerClient.scala

someProducer !! "message" 

SomeProducerException.scala

class SomeProducer extends Actor with Producer {
  // ...
  override def receiveAfterProduce = {
    case Failure(cause, _) => for (sf <- self.senderFuture) sf.completeWithException(cause)
    case message           => self.reply(message)  
  }
}

SupervisedFileConsumer.scala

class SupervisedFileConsumer extends Actor with Consumer {
  override def autoack = false

  def endpointUri = "file:data/input/actor?delete=true"

  def receive = {
    case msg: Message => {
      // do something that may throw an exception
      // ...

      // positively acknowledge receipt of message
      self.reply(Ack)
    }
  }

  override def preRestart(reason: scala.Throwable) {
    self.reply_?(Failure(reason))
  }

  override def postStop() {
    self.reply_?(Ack) // or reply with a Failure to
                      // keep the corresponding file
  }
}

SupervisedProducer.scala

class SupervisedProducer extends Actor with Producer with ChannelManagement {
  def endpointUri = ...

  // manage a replyChannel for the default receive implementation
  override def receive = manageReplyChannelFor(super.receive)

  override def receiveAfterProduce = {
    // exception thrown here on failure, for example
    // ...
  }
  override def preRestartProducer(reason: Throwable) {
    for (c <- replyChannel) c ! Failure(reason)
  }
  override def postStop() {
    for (c <- replyChannel) c ! Failure(new Exception("actor stopped by supervisor"))
  }
}

SupervisedProducerNew.scala

class SupervisedProducer extends Actor with Producer {
  def endpointUri = ...

  override def receiveAfterProduce = {
    // exception thrown here on failure, for example
    // ...
  }
  override def preRestartProducer(reason: Throwable) {
    self.reply_?(Failure(reason))
  }
  override def postStop() {
    self.reply_?(Failure(new Exception("actor stopped by supervisor")))
  }
}