akka-camel producer/consumer

gistfile1.scala

import akka.camel.{Consumer, CamelMessage}
import akka.actor.{Props, ActorSystem}
import akka.util.Timeout
import akka.util.duration._
import akka.pattern.ask
import scaladays2012.{Email, EmailerConfig, Emailer}

class HttpConsumerExample extends Consumer{
  def endpointUri = "jetty://http://0.0.0.0:1111/demo"

  implicit val timeout = Timeout(30 seconds)
  val emailer = context.actorOf(Props(new Emailer(EmailerConfig(System.getProperty("gmail.password")))), "Emailer")

  protected def receive = {
    case msg : CamelMessage => {
      val name = msg.header("name").getOrElse("Stranger")
      val message = "Hello %s from ScalaDays 2012!" format name
      sender ! message
      for( email <- msg header "email" ) {
        emailer ? Email("[email protected]", email.toString, message, "hello") // ignoring the response
      }
    }
  }
}

import akka.actor.Actor
import akka.camel.{Failure, CamelMessage, Producer}

case class EmailerConfig(gmailPassword: Required[String])
case class Email(from: String, to: String, subject: String, body: String)

class Emailer(cfg: EmailerConfig) extends Actor with Producer{
  def endpointUri = "smtps://smtp.gmail.com:[email protected]&password=%s&debugMode=false&defaultEncoding=utf-8" format cfg.gmailPassword.value


  override protected def transformOutgoingMessage(msg: Any) = msg match {
    case Email(from, to, subject, body) =>
      new CamelMessage(body, Map("from" -> from,  "to" -> to,  "subject"-> subject))
  }

  override protected def transformResponse(msg : Any) = msg match {
    case resp: Failure => akka.actor.Status.Failure(resp.getCause)
    case _ => msg
  }

}


object HttpConsumerApp extends App {
  val sys = ActorSystem("test")
  sys.actorOf(Props[HttpConsumerExample], "HttpConsumer")
  sys.awaitTermination()
}