MsgConsumer.scala

import akka.actor.ActorRef
import com.codahale.jerkson.Json.parse
import com.codahale.jerkson.ParsingException
import com.rabbitmq.client.{Channel, DefaultConsumer, Envelope}
import com.rabbitmq.client.AMQP.BasicProperties


class MsgConsumer(channel: Channel, target: ActorRef) extends DefaultConsumer(channel) {

  override def handleDelivery(consumer_tag: String,
                              envelope: Envelope,
                              properties: BasicProperties,
                              body: Array[Byte])
  {
    val delivery_tag = envelope.getDeliveryTag
    val body_text = new String(body)
    try {
      val msg = parse[Map[String, Any]](body_text)
      target ! msg
      channel.basicAck(delivery_tag, false)
    } catch {
      case e: ParsingException => {
        channel.basicReject(delivery_tag, false)
      }
    }
  }

}